class WSSubject(Observer):
    def __init__(self, web_socket: WebSocketResponse):
        super(WSSubject, self).__init__()
        self._web_socket = web_socket
        self._push_subject = Subject()

    def to_observable(self):
        return self._push_subject

    async def process(self):
        async for msg in self._web_socket:
            self._push_subject.on_next(msg)

        self._push_subject.on_completed()

    def on_next(self, data):
        self._web_socket.send_str(data)

    def on_completed(self):
        # close web socket
        # has to be coroutine to close ws
        pass

    def on_error(self, error):
        # send error and close web socket
        pass

def setup_streams(store, debug=False):
    reducer = reduce_action

    if debug:
        reducer = debug_reducer(print_traceback=False)(reducer)

    action_stream = Subject() \
        .merge(*state_mutating_actions) \
        .map(partial(resolve_action, store=store)) \
        .filter(lambda action: action is not None) \
        .map(lambda action: reducer(store.value, action))
    action_stream.subscribe(store)

    changed_notes_stream = store \
        .map(selectors.current_note) \
        .distinct_until_changed() \
        .debounce(1000, scheduler=scheduler) \
        .pausable_buffered(note_ready_stream)
    changed_notes_stream.subscribe(do_save_note)

    ui_state_stream = store \
        .map(selectors.ui_state) \
        .debounce(500, scheduler=scheduler) \
        .distinct_until_changed()
    ui_state_stream.subscribe(save_ui_state)

    actions.quit.subscribe(do_quit)

class WSHandler(WebSocketHandler):
    def open(self):
        print("WebSocket opened")

        # A Subject is both an observable and observer, so we can both subscribe
        # to it and also feed (on_next) it with new values
        self.subject = Subject()

        # Now we take on our magic glasses and project the stream of bytes into
        # a ...
        query = self.subject.map(
                lambda obj: obj["keycode"] # 1. stream of keycodes
            ).window_with_count(
                10, 1 # 2. stream of windows (10 ints long)
            ).select_many(
                # 3. stream of booleans, True or False
                lambda win: win.sequence_equal(codes)
            ).filter(
                lambda equal: equal # 4. stream of Trues
            )
        # 4. we then subscribe to the Trues, and signal Konami! if we see any
        query.subscribe(lambda x: self.write_message("Konami!"))

    def on_message(self, message):
        obj = json_decode(message)
        self.subject.on_next(obj)

    def on_close(self):
        print("WebSocket closed")

class WSHandler(WebSocketHandler):
    def open(self):
        print("WebSocket opened")

        # A Subject is both an observable and observer, so we can both subscribe
        # to it and also feed (on_next) it with new values
        self.stream = Subject()

        # Get all distinct key up events from the input and only fire if long enough and distinct
        query = (
            self.stream.map(lambda x: x["term"])
            .filter(lambda text: len(text) > 2)  # Only if the text is longer than 2 characters
            .debounce(0.750, scheduler=scheduler)  # Pause for 750ms
            .distinct_until_changed()
        )  # Only if the value has changed

        searcher = query.flat_map_latest(search_wikipedia)

        def send_response(x):
            self.write_message(x.body)

        def on_error(ex):
            print(ex)

        searcher.subscribe(send_response, on_error)

    def on_message(self, message):
        obj = json_decode(message)
        self.stream.on_next(obj)

    def on_close(self):
        print("WebSocket closed")

def test_accepts_multiple_subscription_fields_defined_in_schema():
    SubscriptionTypeMultiple = GraphQLObjectType(
        name='Subscription',
        fields=OrderedDict([
            ('importantEmail', GraphQLField(EmailEventType)),
            ('nonImportantEmail', GraphQLField(EmailEventType)),
        ])
    )
    test_schema = GraphQLSchema(
        query=QueryType,
        subscription=SubscriptionTypeMultiple
    )

    stream = Subject()
    send_important_email, subscription = create_subscription(
        stream, test_schema)

    email = Email(
        from_='[email protected]',
        subject='Alright',
        message='Tests are good',
        unread=True,
    )
    l = []
    stream.subscribe(l.append)
    send_important_email(email)
    assert l[0][0] == email

class Node(object):
    def __init__(self, node_dict):
        self.data = Subject()
        self.attrs = node_dict

    def send_to(self, to_node):
        self.data.subscribe(to_node.on_data)

    def compute_output(self, data):
        return self.transform(data, self.attrs.get('transformAttributes', {}))

    def set_transform(self, transform):
        self.transform = transform
        if self._data is not None:
            self.data.on_next(self.compute_output(self._data))

    @staticmethod
    def create(node_dict):
        from .data import DataNode
        from .transform import TransformNode
        from .sink import SinkNode

        node_type_map = {
            'DATA_NODE': DataNode,
            'TRANSFORM_NODE': TransformNode,
            'SINK_NODE': SinkNode
        }

        node_type = node_dict['nodeType']
        return node_type_map[node_type](node_dict)

class PausableObservable(ObservableBase):
    def __init__(self, source, pauser=None):
        self.source = source
        self.controller = Subject()

        if pauser and hasattr(pauser, "subscribe"):
            self.pauser = self.controller.merge(pauser)
        else:
            self.pauser = self.controller

        super(PausableObservable, self).__init__()

    def _subscribe_core(self, observer):
        conn = self.source.publish()
        subscription = conn.subscribe(observer)
        connection = [Disposable.empty()]

        def on_next(b):
            if b:
                connection[0] = conn.connect()
            else:
                connection[0].dispose()
                connection[0] = Disposable.empty()

        pausable = self.pauser.distinct_until_changed().subscribe(on_next)
        return CompositeDisposable(subscription, connection[0], pausable)

    def pause(self):
        self.controller.on_next(False)

    def resume(self):
        self.controller.on_next(True)

    def get_subject(self):
        subject = Subject()

        subject.subscribe(
            self.on_next,
            self.on_error,
            self.on_completed
        )

        return subject

def build_threads(stream_to_testers, thread_target=_perform_subscribe):
    threads = []

    for stream, testers in stream_to_testers.items():
        subject = Subject()
        for tester in testers:
            subject.subscribe(tester)

        threads.append(_make_thread(thread_target, stream, subject))

    return threads

class Pirs:
    def __init__(self):
        io.setmode(io.BCM)
        self.pir_pin = 4
        io.setup(self.pir_pin, io.IN)
        self.pirStream = Subject()
        io.add_event_detect(self.pir_pin, io.RISING, callback=self.hit_callback)

    def hit_callback(self, channel):
        logger.logger.debug('PIR DETECTION!')
        self.pirStream.on_next(True)

class Window(Gtk.Window):
    def __init__(self):
        super().__init__()
        self.resize(600, 600)

        self.add_events(Gdk.EventMask.POINTER_MOTION_MASK)
        self.connect("motion-notify-event", self.on_mouse_move)

        self.mousemove = Subject()

    def on_mouse_move(self, widget, event):
        self.mousemove.on_next((event.x, event.y))

            def on_next_left(value):
                s = Subject()

                with self.lock:
                    _id = left_id[0]
                    left_id[0] += 1
                    left_map[_id] = s

                try:
                    result = result_selector(value, add_ref(s, r))
                except Exception as e:
                    log.error("*** Exception: %s" % e)
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)
                    return

                observer.on_next(result)

                for right_value in right_map.values():
                    s.on_next(right_value)

                md = SingleAssignmentDisposable()
                group.add(md)

                def expire():
                    if _id in left_map:
                        del left_map[_id]
                        s.on_completed()

                    group.remove(md)

                try:
                    duration = left_duration_selector(value)
                except Exception as e:
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)
                    return

                def on_error(e):
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)

                md.disposable = duration.take(1).subscribe(
                    nothing,
                    on_error,
                    expire)

                def action(scheduler, state=None):
                    s = None

                    if is_shift:
                        s = Subject()
                        q.append(s)
                        observer.on_next(add_ref(s, ref_count_disposable))

                    if is_span:
                        s = q.pop(0)
                        s.on_completed()

                    create_timer()

class Window(QWidget):

    def __init__(self):
        super(QWidget, self).__init__()
        self.setWindowTitle("Rx for Python rocks")
        self.resize(600, 600)
        self.setMouseTracking(True)

        # This Subject is used to transmit mouse moves to labels
        self.mousemove = Subject()

    def mouseMoveEvent(self, event):
        self.mousemove.on_next((event.x(), event.y()))

class Frame(wx.Frame):

    def __init__(self):
        super(Frame, self).__init__(None)
        self.SetTitle("Rx for Python rocks")
        self.SetSize((600, 600))

        # This Subject is used to transmit mouse moves to labels
        self.mousemove = Subject()

        self.Bind(wx.EVT_MOTION, self.OnMotion)

    def OnMotion(self, event):
        self.mousemove.on_next((event.GetX(), event.GetY()))

class DoorListener:
    def __init__(self):
        self.door_message_server = MessageServer(config.door_tag_port)
        self.openDoorStream = Subject()
        self.vibeDoorStream = Subject()

    def on_message(self, data):
        if data == 'door opened':
            self.openDoorStream.on_next(True)
        if data == 'door hit':
            self.vibeDoorStream.on_next(True)

    def listen(self):
        self.door_message_server.listen(self.on_message)

            def on_next_left(value):
                subject = Subject()

                with left.lock:
                    _id = left_id[0]
                    left_id[0] += 1
                    left_map[_id] = subject

                try:
                    result = (value, add_ref(subject, rcd))
                except Exception as e:
                    log.error("*** Exception: %s" % e)
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)
                    return

                observer.on_next(result)

                for right_value in right_map.values():
                    subject.on_next(right_value)

                md = SingleAssignmentDisposable()
                group.add(md)

                def expire():
                    if _id in left_map:
                        del left_map[_id]
                        subject.on_completed()

                    group.remove(md)

                try:
                    duration = left_duration_mapper(value)
                except Exception as e:
                    for left_value in left_map.values():
                        left_value.on_error(e)

                    observer.on_error(e)
                    return

                def on_error(error):
                    for left_value in left_map.values():
                        left_value.on_error(error)

                    observer.on_error(error)

                md.disposable = duration.pipe(ops.take(1)).subscribe_(nothing, on_error, expire, scheduler)

    def __init__(self, new_players, exiting_players):
        self._new_players = []
        self._players = []
        self._removed_players = []

        new_players.subscribe(self.add_player)
        exiting_players.subscribe(self.remove_player)

        # streams api
        self.new_players_broadcast = Subject()
        self.removed_players_broadcast = Subject()
        self.collisions = Subject()
        self.players = BehaviorSubject([])
        self.players_count = self.players \
            .map(lambda ps: len(ps))

    def open(self):
        print("WebSocket opened")

        # Subject одновременно и observable, и observer
        self.subject = Subject()

        def send_response(x):
            self.write_message(json.dumps(x))

        def on_error(ex):
            print(ex)

        user_input = self.subject.throttle_last(
            1000  # На заданном временном промежутке получать последнее значение
        ).start_with(
            ''  # Сразу же после подписки отправляет значение по умолчанию
        ).filter(
            lambda text: not text or len(text) > 2
        )

        interval_obs = Observable.interval(
            60000  # Отдает значение раз в 60с (для периодического обновления)
        ).start_with(0)

        # combine_latest собирает 2 потока из запросов пользователя и временных
        # интервалов, срабатывает на любое сообщение из каждого потока
        self.combine_latest_sbs = user_input.combine_latest(
            interval_obs, lambda input_val, i: input_val
        ).do_action(  # Срабатывает на каждый выпущенный элемент
            # Отправляет сообщение для очистки списка на фронтэнд
            lambda x: send_response('clear')
        ).flat_map(
            # В цепочку встраивается observable для получения списка
            self.get_data
        ).subscribe(send_response, on_error)

    def open(self):
        scheduler = AsyncIOScheduler()

        print("WebSocket opened")

        # A Subject is both an observable and observer, so we can both subscribe
        # to it and also feed (send) it with new values
        self.subject = Subject()

        # Get all distinct key up events from the input and only fire if long enough and distinct
        searcher = self.subject.pipe(
            ops.map(lambda x: x["term"]),
            ops.filter(lambda text: len(text) > 2),  # Only if the text is longer than 2 characters
            ops.debounce(0.750),                     # Pause for 750ms
            ops.distinct_until_changed(),            # Only if the value has changed
            ops.flat_map_latest(search_wikipedia)
        )

        def send_response(x):
            self.write_message(x.body)

        def on_error(ex):
            print(ex)

        searcher.subscribe(send_response, on_error, scheduler=scheduler)