@Test public void testValidClient() { RadixNode node = mock(RadixNode.class); WebSocketClient ws = mock(WebSocketClient.class); when(ws.sendMessage(any())).thenReturn(true); FindANodeRequestAction request = mock(FindANodeRequestAction.class); FindANodeEpic findANodeFunction = new FindANodeEpic(new GetFirstSelector()); TestObserver<RadixNodeAction> testObserver = TestObserver.create(); findANodeFunction.epic( Observable.<RadixNodeAction>just(request).concatWith(Observable.never()), Observable.just(RadixNetworkState.of(node, mockedNodeState(WebSocketStatus.CONNECTED))) ) .subscribe(testObserver); testObserver.awaitCount(1); testObserver.assertValue(u -> u.getNode().equals(node)); }
@Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable) { return actions.ofType(FindANodeRequestAction.class) .flatMap(a -> { Observable<List<RadixNode>> connectedNodes = stateObservable .map(FindANodeEpic::getConnectedNodes) .replay(1) .autoConnect(2); Observable<RadixNodeAction> selectedNode = connectedNodes .filter(viablePeerList -> !viablePeerList.isEmpty()) .firstOrError() .map(selector::apply) .<RadixNodeAction>map(n -> new FindANodeResultAction(n, a)) .cache() .toObservable(); Observable<RadixNodeAction> findConnectionActionsStream = connectedNodes .filter(List::isEmpty) .firstOrError() .ignoreElement() .andThen( Observable .interval(0, NEXT_CONNECTION_THROTTLE_TIMEOUT_SECS, TimeUnit.SECONDS) .withLatestFrom(stateObservable, (i, s) -> s) .flatMapIterable(this::nextConnectionRequest) ) .takeUntil(selectedNode) .replay(1) .autoConnect(2); Observable<RadixNodeAction> cleanupConnections = findConnectionActionsStream .ofType(ConnectWebSocketAction.class) .flatMap(c -> { final RadixNode node = c.getNode(); return selectedNode .map(RadixNodeAction::getNode) .filter(selected -> !node.equals(selected)) .map(i -> CloseWebSocketAction.of(node)); }); return findConnectionActionsStream.concatWith(selectedNode).mergeWith(cleanupConnections); }); }
FindANodeEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable) { return actions.ofType(FindANodeRequestAction.class) .flatMap(a -> { Observable<List<RadixNode>> connectedNodes = stateObservable .map(FindANodeEpic::getConnectedNodes) .replay(1) .autoConnect(2); Observable<RadixNodeAction> selectedNode = connectedNodes .filter(viablePeerList -> !viablePeerList.isEmpty()) .firstOrError() .map(selector::apply) .<RadixNodeAction>map(n -> new FindANodeResultAction(n, a)) .cache() .toObservable(); Observable<RadixNodeAction> findConnectionActionsStream = connectedNodes .filter(List::isEmpty) .firstOrError() .ignoreElement() .andThen( Observable .interval(0, NEXT_CONNECTION_THROTTLE_TIMEOUT_SECS, TimeUnit.SECONDS) .withLatestFrom(stateObservable, (i, s) -> s) .flatMapIterable(this::nextConnectionRequest) ) .takeUntil(selectedNode) .replay(1) .autoConnect(2); Observable<RadixNodeAction> cleanupConnections = findConnectionActionsStream .ofType(ConnectWebSocketAction.class) .flatMap(c -> { final RadixNode node = c.getNode(); return selectedNode .map(RadixNodeAction::getNode) .filter(selected -> !node.equals(selected)) .map(i -> CloseWebSocketAction.of(node)); }); return findConnectionActionsStream.concatWith(selectedNode).mergeWith(cleanupConnections); }); } }
FindANodeEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable) { return actions.ofType(FindANodeRequestAction.class) .flatMap(a -> { Observable<List<RadixNode>> connectedNodes = stateObservable .map(FindANodeEpic::getConnectedNodes) .replay(1) .autoConnect(2); Observable<RadixNodeAction> selectedNode = connectedNodes .filter(viablePeerList -> !viablePeerList.isEmpty()) .firstOrError() .map(selector::apply) .<RadixNodeAction>map(n -> new FindANodeResultAction(n, a)) .cache() .toObservable(); Observable<RadixNodeAction> findConnectionActionsStream = connectedNodes .filter(List::isEmpty) .firstOrError() .ignoreElement() .andThen( Observable .interval(0, NEXT_CONNECTION_THROTTLE_TIMEOUT_SECS, TimeUnit.SECONDS) .withLatestFrom(stateObservable, (i, s) -> s) .flatMapIterable(this::nextConnectionRequest) ) .takeUntil(selectedNode) .replay(1) .autoConnect(2); Observable<RadixNodeAction> cleanupConnections = findConnectionActionsStream .ofType(ConnectWebSocketAction.class) .flatMap(c -> { final RadixNode node = c.getNode(); return selectedNode .map(RadixNodeAction::getNode) .filter(selected -> !node.equals(selected)) .map(i -> CloseWebSocketAction.of(node)); }); return findConnectionActionsStream.concatWith(selectedNode).mergeWith(cleanupConnections); }); } FindANodeEpic(RadixPeerSelector selector); }
FindANodeEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable) { return actions.ofType(FindANodeRequestAction.class) .flatMap(a -> { Observable<List<RadixNode>> connectedNodes = stateObservable .map(FindANodeEpic::getConnectedNodes) .replay(1) .autoConnect(2); Observable<RadixNodeAction> selectedNode = connectedNodes .filter(viablePeerList -> !viablePeerList.isEmpty()) .firstOrError() .map(selector::apply) .<RadixNodeAction>map(n -> new FindANodeResultAction(n, a)) .cache() .toObservable(); Observable<RadixNodeAction> findConnectionActionsStream = connectedNodes .filter(List::isEmpty) .firstOrError() .ignoreElement() .andThen( Observable .interval(0, NEXT_CONNECTION_THROTTLE_TIMEOUT_SECS, TimeUnit.SECONDS) .withLatestFrom(stateObservable, (i, s) -> s) .flatMapIterable(this::nextConnectionRequest) ) .takeUntil(selectedNode) .replay(1) .autoConnect(2); Observable<RadixNodeAction> cleanupConnections = findConnectionActionsStream .ofType(ConnectWebSocketAction.class) .flatMap(c -> { final RadixNode node = c.getNode(); return selectedNode .map(RadixNodeAction::getNode) .filter(selected -> !node.equals(selected)) .map(i -> CloseWebSocketAction.of(node)); }); return findConnectionActionsStream.concatWith(selectedNode).mergeWith(cleanupConnections); }); } FindANodeEpic(RadixPeerSelector selector); @Override Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable); }
FindANodeEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable) { return actions.ofType(FindANodeRequestAction.class) .flatMap(a -> { Observable<List<RadixNode>> connectedNodes = stateObservable .map(FindANodeEpic::getConnectedNodes) .replay(1) .autoConnect(2); Observable<RadixNodeAction> selectedNode = connectedNodes .filter(viablePeerList -> !viablePeerList.isEmpty()) .firstOrError() .map(selector::apply) .<RadixNodeAction>map(n -> new FindANodeResultAction(n, a)) .cache() .toObservable(); Observable<RadixNodeAction> findConnectionActionsStream = connectedNodes .filter(List::isEmpty) .firstOrError() .ignoreElement() .andThen( Observable .interval(0, NEXT_CONNECTION_THROTTLE_TIMEOUT_SECS, TimeUnit.SECONDS) .withLatestFrom(stateObservable, (i, s) -> s) .flatMapIterable(this::nextConnectionRequest) ) .takeUntil(selectedNode) .replay(1) .autoConnect(2); Observable<RadixNodeAction> cleanupConnections = findConnectionActionsStream .ofType(ConnectWebSocketAction.class) .flatMap(c -> { final RadixNode node = c.getNode(); return selectedNode .map(RadixNodeAction::getNode) .filter(selected -> !node.equals(selected)) .map(i -> CloseWebSocketAction.of(node)); }); return findConnectionActionsStream.concatWith(selectedNode).mergeWith(cleanupConnections); }); } FindANodeEpic(RadixPeerSelector selector); @Override Observable<RadixNodeAction> epic(Observable<RadixNodeAction> actions, Observable<RadixNetworkState> stateObservable);  }