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);  }
@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)); }
@Test public void disconnectedNodeConnectionTest() { RadixNode node = mock(RadixNode.class); FindANodeEpic findANodeEpic = new FindANodeEpic(new RandomSelector()); RadixNodeState nodeState = mock(RadixNodeState.class); when(nodeState.getStatus()).thenReturn(WebSocketStatus.DISCONNECTED); FindANodeRequestAction request = mock(FindANodeRequestAction.class); TestObserver<RadixNodeAction> testObserver = TestObserver.create(); findANodeEpic.epic( Observable.<RadixNodeAction>just(request).concatWith(Observable.never()), Observable.just(RadixNetworkState.of(node, nodeState)).concatWith(Observable.never()) ) .subscribe(testObserver); testObserver.awaitCount(1); testObserver.assertValue(u -> u instanceof ConnectWebSocketAction); testObserver.assertValue(u -> u.getNode().equals(node)); testObserver.assertNotComplete(); }
@Test public void dontConnectToAllNodesTest() { RadixNode connectedPeer = mock(RadixNode.class); Map<RadixNode, RadixNodeState> networkStateMap = IntStream.range(0, 100).boxed().collect(Collectors.toMap( i -> { if (i == 0) { return connectedPeer; } RadixNode peer = mock(RadixNode.class); return peer; }, i -> mockedNodeState(i == 0 ? WebSocketStatus.CONNECTED : WebSocketStatus.DISCONNECTED) )); 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(new RadixNetworkState(networkStateMap)).concatWith(Observable.never()) ) .subscribe(testObserver); testObserver.assertValue(u -> u.getNode().equals(connectedPeer)); testObserver.assertValue(u -> ((FindANodeResultAction) u).getRequest().equals(request)); }
@Test public void when_first_node_takes_too_long__then_second_node_should_connect() { RadixNode badPeer = mock(RadixNode.class); RadixNode goodPeer = mock(RadixNode.class); ReplaySubject<RadixNetworkState> networkState = ReplaySubject.create(); networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.DISCONNECTED), goodPeer, mockedNodeState(WebSocketStatus.DISCONNECTED) ))); FindANodeEpic findANodeEpic = new FindANodeEpic(new GetFirstSelector()); TestObserver<RadixNodeAction> testObserver = TestObserver.create(); FindANodeRequestAction request = mock(FindANodeRequestAction.class); findANodeEpic.epic( Observable.<RadixNodeAction>just(request).concatWith(Observable.never()), networkState ) .doOnNext(i -> { if (i.getNode().equals(badPeer)) { networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.CONNECTING), goodPeer, mockedNodeState(WebSocketStatus.DISCONNECTED) ))); } else { networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.CONNECTING), goodPeer, mockedNodeState(WebSocketStatus.CONNECTED) ))); } }) .subscribe(testObserver); testObserver.awaitCount(4); testObserver.assertValueAt(0, u -> u instanceof ConnectWebSocketAction && u.getNode().equals(badPeer)); testObserver.assertValueAt(1, u -> u instanceof ConnectWebSocketAction && u.getNode().equals(goodPeer)); testObserver.assertValueAt(2, u -> u instanceof CloseWebSocketAction && u.getNode().equals(badPeer)); testObserver.assertValueAt(3, u -> ((FindANodeResultAction) u).getRequest().equals(request) && u.getNode().equals(goodPeer)); }
@Test public void when_first_node_fails__then_second_node_should_connect() { RadixNode badPeer = mock(RadixNode.class); RadixNode goodPeer = mock(RadixNode.class); ReplaySubject<RadixNetworkState> networkState = ReplaySubject.create(); networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.DISCONNECTED), goodPeer, mockedNodeState(WebSocketStatus.DISCONNECTED) ))); FindANodeEpic findANodeEpic = new FindANodeEpic(new GetFirstSelector()); TestObserver<RadixNodeAction> testObserver = TestObserver.create(); FindANodeRequestAction request = mock(FindANodeRequestAction.class); findANodeEpic.epic( Observable.<RadixNodeAction>just(request).concatWith(Observable.never()), networkState ) .doOnNext(i -> { if (i.getNode().equals(badPeer)) { networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.FAILED), goodPeer, mockedNodeState(WebSocketStatus.DISCONNECTED) ))); } else { networkState.onNext(new RadixNetworkState(ImmutableMap.of( badPeer, mockedNodeState(WebSocketStatus.FAILED), goodPeer, mockedNodeState(WebSocketStatus.CONNECTED) ))); } }) .subscribe(testObserver); testObserver.awaitCount(4); testObserver.assertValueAt(0, u -> u instanceof ConnectWebSocketAction && u.getNode().equals(badPeer)); testObserver.assertValueAt(1, u -> u instanceof ConnectWebSocketAction && u.getNode().equals(goodPeer)); testObserver.assertValueAt(2, u -> u instanceof CloseWebSocketAction && u.getNode().equals(badPeer)); testObserver.assertValueAt(3, u -> ((FindANodeResultAction) u).getRequest().equals(request) && u.getNode().equals(goodPeer)); }