How to use the moler.threaded_moler_connection.ThreadedMolerConnection function in moler

def test_can_open_and_close_connection(tcp_connection_class,
                                       integration_tcp_server_and_pipe):
    """
    Not so atomic test (checks 2 things) but:
    - it is integration tests
    - anyway open needs close as cleanup to not have resources leaking in tests
    """
    from moler.threaded_moler_connection import ThreadedMolerConnection
    (tcp_server, tcp_server_pipe) = integration_tcp_server_and_pipe

    moler_conn = ThreadedMolerConnection()
    connection = tcp_connection_class(moler_connection=moler_conn, port=tcp_server.port, host=tcp_server.host)
    connection.open()
    connection.close()
    time.sleep(0.1)  # otherwise we have race between server's pipe and from-client-connection
    tcp_server_pipe.send(("get history", {}))
    dialog_with_server = tcp_server_pipe.recv()
    assert 'Client connected' in dialog_with_server
    assert 'Client disconnected' in dialog_with_server

def connection_observer(observer_runner):
    from moler.threaded_moler_connection import ThreadedMolerConnection
    moler_conn = ThreadedMolerConnection()
    observer = NetworkDownDetector(connection=moler_conn, runner=observer_runner)
    yield observer
    # remove exceptions collected inside ConnectionObserver
    ConnectionObserver.get_unraised_exceptions(remove=True)

def test_can_open_and_close_connection_as_context_manager(tcp_connection_class,
                                                          integration_tcp_server_and_pipe):
    from moler.threaded_moler_connection import ThreadedMolerConnection
    (tcp_server, tcp_server_pipe) = integration_tcp_server_and_pipe

    moler_conn = ThreadedMolerConnection()
    connection = tcp_connection_class(moler_connection=moler_conn, port=tcp_server.port, host=tcp_server.host)
    with connection.open():
        pass
    dialog_with_server = _wait_for_last_message(tcp_server_pipe=tcp_server_pipe, last_message='Client disconnected',
                                                timeout=5)
    assert 'Client connected' in dialog_with_server
    assert 'Client disconnected' in dialog_with_server

def test_can_send_binary_data_over_connection(tcp_connection_class,
                                              integration_tcp_server_and_pipe):
    from moler.threaded_moler_connection import ThreadedMolerConnection
    (tcp_server, tcp_server_pipe) = integration_tcp_server_and_pipe

    moler_conn = ThreadedMolerConnection()  # no decoder, just pass bytes 1:1
    connection = tcp_connection_class(moler_connection=moler_conn, port=tcp_server.port, host=tcp_server.host)
    with connection.open():
        moler_conn.send(data=b'data to be send')  # TODO: await moler_conn.send(data=b'data to be send') ???
        time.sleep(0.1)  # otherwise we have race between server's pipe and from-client-connection
        tcp_server_pipe.send(("get history", {}))
        dialog_with_server = tcp_server_pipe.recv()
        assert ['Received data:', b'data to be send'] == dialog_with_server[-1]

def failing_net_down_detector(fail_on_data, fail_by_raising, runner):
    from moler.threaded_moler_connection import ThreadedMolerConnection

    class FailingNetworkDownDetector(NetworkDownDetector):
        def data_received(self, data, recv_time):
            if data == fail_on_data:
                raise fail_by_raising
            return super(FailingNetworkDownDetector, self).data_received(data, recv_time)

    moler_conn = ThreadedMolerConnection()
    failing_detector = FailingNetworkDownDetector(connection=moler_conn, runner=runner)
    yield failing_detector
    # remove exceptions collected inside ConnectionObserver
    ConnectionObserver.get_unraised_exceptions(remove=True)

def conn_observer(request):
    moler_conn = ThreadedMolerConnection(how2send=mock.MagicMock())
    if request.param == 'generic_observer':
        observer = NetworkDownDetector(connection=moler_conn)
    elif request.param == 'event':
        observer = MyEvent(connection=moler_conn)
    elif request.param == 'command':
        observer = MyCommand(connection=moler_conn)
    return observer

def test_events_true_all():
    connection = ThreadedMolerConnection()
    events = list()
    patterns = ("aaa", "bbb")
    for pattern in patterns:
        event = Wait4prompt(connection=connection, till_occurs_times=1, prompt=pattern)
        event.start()
        events.append(event)
        connection.data_received(pattern, datetime.datetime.now())
    assert EventAwaiter.wait_for_all(timeout=0.1, events=events) is True
    done, not_done = EventAwaiter.separate_done_events(events)
    assert 2 == len(done)
    assert 0 == len(not_done)
    EventAwaiter.cancel_all_events(events)

async def ping_observing_task(address):
    logger = logging.getLogger('moler.user.app-code')

    # Lowest layer of Moler's usage (you manually glue all elements):
    # 1. create observer
    net_down_detector = NetworkDownDetector('10.0.2.15')
    # 2. ThreadedMolerConnection is a proxy-glue between observer (speaks str)
    #                                   and curio-connection (speaks bytes)
    moler_conn = ThreadedMolerConnection(decoder=lambda data: data.decode("utf-8"))
    # 3a. glue from proxy to observer
    moler_conn.subscribe(net_down_detector.data_received)

    logger.debug('waiting for data to observe')
    async with curio.meta.finalize(tcp_connection(address)) as tcp_conn:
        async for connection_data in tcp_conn:
            # 3b. glue to proxy from external-IO (curio tcp client connection)
            #    (client has to pass it's received data into Moler's connection)
            moler_conn.data_received(connection_data)
            # 4. Moler's client code must manually check status of observer ...
            if net_down_detector.done():
                # 5. ... to know when it can ask for result
                net_down_time = net_down_detector.result()
                timestamp = time.strftime("%H:%M:%S",
                                          time.localtime(net_down_time))
                logger.debug('Network is down from {}'.format(timestamp))

def main(connections2observe4ip):
    # Starting the clients
    connections = []
    for address, ping_ip in connections2observe4ip:
        host, port = address
        # 1. create Moler's connection that knows encoding
        decoder = lambda data: data.decode("utf-8")
        moler_conn = ThreadedMolerConnection(decoder=decoder)
        # 2. create external-IO connection gluing to Moler's connection
        conn_logger_name = 'threaded.tcp-connection({}:{})'.format(*address)
        conn_logger = logging.getLogger(conn_logger_name)
        tcp_connection = tcp.ThreadedTcp(moler_connection=moler_conn,
                                         port=port, host=host,
                                         logger=conn_logger)
        client_thread = threading.Thread(target=ping_observing_task,
                                         args=(tcp_connection, ping_ip))
        client_thread.start()
        connections.append(client_thread)
    # await observers job to be done
    for client_thread in connections:
        client_thread.join()

- have a means for sending outgoing data via external IO
- have a means for receiving incoming data from external IO
- perform data encoding/decoding to let external IO use pure bytes
- have a means allowing multiple observers to get it's received data (data dispatching)
"""

__author__ = 'Grzegorz Latuszek, Marcin Usielski, Michal Ernst'
__copyright__ = 'Copyright (C) 2018-2020, Nokia'
__email__ = 'grzegorz.latuszek@nokia.com, marcin.usielski@nokia.com, michal.ernst@nokia.com'


from moler.threaded_moler_connection import ThreadedMolerConnection
import logging


class ObservableConnection(ThreadedMolerConnection):

    def __init__(self, *args, **kwargs):
        """
        Create Connection via registering external-IO

        :param how2send: any callable performing outgoing IO
        :param encoder: callable converting data to bytes
        :param decoder: callable restoring data from bytes
        :param name: name assigned to connection
        :param logger_name: take that logger from logging

        Logger is retrieved by logging.getLogger(logger_name)
        If logger_name == "" - take logger "moler.connection."
        If logger_name is None - don't use logging
        """
        super(ObservableConnection, self).__init__(*args, **kwargs)