@Test public void when_live_peers_result_is_observed__add_node_events_should_be_emitted() { RadixNode node = mock(RadixNode.class); when(node.getPort()).thenReturn(12345); Observable<RadixNode> seeds = Observable.just(node); RadixUniverseConfig universe = mock(RadixUniverseConfig.class); ReplaySubject<RadixNodeAction> actions = ReplaySubject.create(); RadixNetworkState networkState = mock(RadixNetworkState.class); when(networkState.getNodeStates()).thenReturn(Collections.emptyMap()); Observable<RadixNetworkState> observableNetworkState = Observable.concat(Observable.just(networkState), Observable.never()); DiscoverNodesEpic discoverNodesEpic = new DiscoverNodesEpic(seeds, universe); Observable<RadixNodeAction> output = discoverNodesEpic.epic(actions, observableNetworkState); TestObserver<RadixNodeAction> testObserver = TestObserver.create(); output.subscribe(testObserver); NodeRunnerData data0 = mock(NodeRunnerData.class); when(data0.getIp()).thenReturn("1.2.3.4"); NodeRunnerData data1 = mock(NodeRunnerData.class); when(data1.getIp()).thenReturn("2.3.4.5"); List<NodeRunnerData> nodeRunnerData = Arrays.asList(data0, data1); actions.onNext(GetLivePeersResultAction.of(node, nodeRunnerData)); testObserver.awaitCount(2); testObserver.assertValueAt(0, a -> a instanceof AddNodeAction); testObserver.assertValueAt(1, a -> a instanceof AddNodeAction); }
@Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState) { Observable<RadixNodeAction> getSeedUniverses = updates .ofType(DiscoverMoreNodesAction.class) .firstOrError() .flatMapObservable(i -> seeds) .<RadixNodeAction>map(GetUniverseRequestAction::of) .onErrorReturn(DiscoverMoreNodesErrorAction::new); Observable<RadixNodeAction> seedUniverseMismatch = updates .ofType(GetUniverseResponseAction.class) .filter(u -> !u.getResult().equals(config)) .map(u -> new NodeUniverseMismatch(u.getNode(), config, u.getResult())); Observable<RadixNode> connectedSeeds = updates .ofType(GetUniverseResponseAction.class) .filter(u -> u.getResult().equals(config)) .map(GetUniverseResponseAction::getNode) .publish() .autoConnect(3); Observable<RadixNodeAction> addSeeds = connectedSeeds.map(AddNodeAction::of); Observable<RadixNodeAction> addSeedData = connectedSeeds.map(GetNodeDataRequestAction::of); Observable<RadixNodeAction> addSeedSiblings = connectedSeeds.map(GetLivePeersRequestAction::of); Observable<RadixNodeAction> addNodes = updates .ofType(GetLivePeersResultAction.class) .flatMap(u -> Observable.combineLatest( Observable.just(u.getResult()), Observable.concat(networkState.firstOrError().toObservable(), Observable.never()), (data, state) -> data.stream() .map(d -> { RadixNode node = new RadixNode(d.getIp(), u.getNode().isSsl(), u.getNode().getPort()); return state.getNodeStates().containsKey(node) ? null : AddNodeAction.of(node, d); }) .filter(Objects::nonNull) .collect(Collectors.toSet()) ).flatMapIterable(i -> i) ); return Observable.merge(Arrays.asList( addSeeds, addSeedData, addSeedSiblings, addNodes, getSeedUniverses, seedUniverseMismatch )); }
DiscoverNodesEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState) { Observable<RadixNodeAction> getSeedUniverses = updates .ofType(DiscoverMoreNodesAction.class) .firstOrError() .flatMapObservable(i -> seeds) .<RadixNodeAction>map(GetUniverseRequestAction::of) .onErrorReturn(DiscoverMoreNodesErrorAction::new); Observable<RadixNodeAction> seedUniverseMismatch = updates .ofType(GetUniverseResponseAction.class) .filter(u -> !u.getResult().equals(config)) .map(u -> new NodeUniverseMismatch(u.getNode(), config, u.getResult())); Observable<RadixNode> connectedSeeds = updates .ofType(GetUniverseResponseAction.class) .filter(u -> u.getResult().equals(config)) .map(GetUniverseResponseAction::getNode) .publish() .autoConnect(3); Observable<RadixNodeAction> addSeeds = connectedSeeds.map(AddNodeAction::of); Observable<RadixNodeAction> addSeedData = connectedSeeds.map(GetNodeDataRequestAction::of); Observable<RadixNodeAction> addSeedSiblings = connectedSeeds.map(GetLivePeersRequestAction::of); Observable<RadixNodeAction> addNodes = updates .ofType(GetLivePeersResultAction.class) .flatMap(u -> Observable.combineLatest( Observable.just(u.getResult()), Observable.concat(networkState.firstOrError().toObservable(), Observable.never()), (data, state) -> data.stream() .map(d -> { RadixNode node = new RadixNode(d.getIp(), u.getNode().isSsl(), u.getNode().getPort()); return state.getNodeStates().containsKey(node) ? null : AddNodeAction.of(node, d); }) .filter(Objects::nonNull) .collect(Collectors.toSet()) ).flatMapIterable(i -> i) ); return Observable.merge(Arrays.asList( addSeeds, addSeedData, addSeedSiblings, addNodes, getSeedUniverses, seedUniverseMismatch )); } }
DiscoverNodesEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState) { Observable<RadixNodeAction> getSeedUniverses = updates .ofType(DiscoverMoreNodesAction.class) .firstOrError() .flatMapObservable(i -> seeds) .<RadixNodeAction>map(GetUniverseRequestAction::of) .onErrorReturn(DiscoverMoreNodesErrorAction::new); Observable<RadixNodeAction> seedUniverseMismatch = updates .ofType(GetUniverseResponseAction.class) .filter(u -> !u.getResult().equals(config)) .map(u -> new NodeUniverseMismatch(u.getNode(), config, u.getResult())); Observable<RadixNode> connectedSeeds = updates .ofType(GetUniverseResponseAction.class) .filter(u -> u.getResult().equals(config)) .map(GetUniverseResponseAction::getNode) .publish() .autoConnect(3); Observable<RadixNodeAction> addSeeds = connectedSeeds.map(AddNodeAction::of); Observable<RadixNodeAction> addSeedData = connectedSeeds.map(GetNodeDataRequestAction::of); Observable<RadixNodeAction> addSeedSiblings = connectedSeeds.map(GetLivePeersRequestAction::of); Observable<RadixNodeAction> addNodes = updates .ofType(GetLivePeersResultAction.class) .flatMap(u -> Observable.combineLatest( Observable.just(u.getResult()), Observable.concat(networkState.firstOrError().toObservable(), Observable.never()), (data, state) -> data.stream() .map(d -> { RadixNode node = new RadixNode(d.getIp(), u.getNode().isSsl(), u.getNode().getPort()); return state.getNodeStates().containsKey(node) ? null : AddNodeAction.of(node, d); }) .filter(Objects::nonNull) .collect(Collectors.toSet()) ).flatMapIterable(i -> i) ); return Observable.merge(Arrays.asList( addSeeds, addSeedData, addSeedSiblings, addNodes, getSeedUniverses, seedUniverseMismatch )); } DiscoverNodesEpic(Observable<RadixNode> seeds, RadixUniverseConfig config); }
DiscoverNodesEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState) { Observable<RadixNodeAction> getSeedUniverses = updates .ofType(DiscoverMoreNodesAction.class) .firstOrError() .flatMapObservable(i -> seeds) .<RadixNodeAction>map(GetUniverseRequestAction::of) .onErrorReturn(DiscoverMoreNodesErrorAction::new); Observable<RadixNodeAction> seedUniverseMismatch = updates .ofType(GetUniverseResponseAction.class) .filter(u -> !u.getResult().equals(config)) .map(u -> new NodeUniverseMismatch(u.getNode(), config, u.getResult())); Observable<RadixNode> connectedSeeds = updates .ofType(GetUniverseResponseAction.class) .filter(u -> u.getResult().equals(config)) .map(GetUniverseResponseAction::getNode) .publish() .autoConnect(3); Observable<RadixNodeAction> addSeeds = connectedSeeds.map(AddNodeAction::of); Observable<RadixNodeAction> addSeedData = connectedSeeds.map(GetNodeDataRequestAction::of); Observable<RadixNodeAction> addSeedSiblings = connectedSeeds.map(GetLivePeersRequestAction::of); Observable<RadixNodeAction> addNodes = updates .ofType(GetLivePeersResultAction.class) .flatMap(u -> Observable.combineLatest( Observable.just(u.getResult()), Observable.concat(networkState.firstOrError().toObservable(), Observable.never()), (data, state) -> data.stream() .map(d -> { RadixNode node = new RadixNode(d.getIp(), u.getNode().isSsl(), u.getNode().getPort()); return state.getNodeStates().containsKey(node) ? null : AddNodeAction.of(node, d); }) .filter(Objects::nonNull) .collect(Collectors.toSet()) ).flatMapIterable(i -> i) ); return Observable.merge(Arrays.asList( addSeeds, addSeedData, addSeedSiblings, addNodes, getSeedUniverses, seedUniverseMismatch )); } DiscoverNodesEpic(Observable<RadixNode> seeds, RadixUniverseConfig config); @Override Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState); }
DiscoverNodesEpic implements RadixNetworkEpic { @Override public Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState) { Observable<RadixNodeAction> getSeedUniverses = updates .ofType(DiscoverMoreNodesAction.class) .firstOrError() .flatMapObservable(i -> seeds) .<RadixNodeAction>map(GetUniverseRequestAction::of) .onErrorReturn(DiscoverMoreNodesErrorAction::new); Observable<RadixNodeAction> seedUniverseMismatch = updates .ofType(GetUniverseResponseAction.class) .filter(u -> !u.getResult().equals(config)) .map(u -> new NodeUniverseMismatch(u.getNode(), config, u.getResult())); Observable<RadixNode> connectedSeeds = updates .ofType(GetUniverseResponseAction.class) .filter(u -> u.getResult().equals(config)) .map(GetUniverseResponseAction::getNode) .publish() .autoConnect(3); Observable<RadixNodeAction> addSeeds = connectedSeeds.map(AddNodeAction::of); Observable<RadixNodeAction> addSeedData = connectedSeeds.map(GetNodeDataRequestAction::of); Observable<RadixNodeAction> addSeedSiblings = connectedSeeds.map(GetLivePeersRequestAction::of); Observable<RadixNodeAction> addNodes = updates .ofType(GetLivePeersResultAction.class) .flatMap(u -> Observable.combineLatest( Observable.just(u.getResult()), Observable.concat(networkState.firstOrError().toObservable(), Observable.never()), (data, state) -> data.stream() .map(d -> { RadixNode node = new RadixNode(d.getIp(), u.getNode().isSsl(), u.getNode().getPort()); return state.getNodeStates().containsKey(node) ? null : AddNodeAction.of(node, d); }) .filter(Objects::nonNull) .collect(Collectors.toSet()) ).flatMapIterable(i -> i) ); return Observable.merge(Arrays.asList( addSeeds, addSeedData, addSeedSiblings, addNodes, getSeedUniverses, seedUniverseMismatch )); } DiscoverNodesEpic(Observable<RadixNode> seeds, RadixUniverseConfig config); @Override Observable<RadixNodeAction> epic(Observable<RadixNodeAction> updates, Observable<RadixNetworkState> networkState);  }