@Test public void deltaTestWithIds() { ReplaySubject<List<Endpoint>> subject = ReplaySubject.create(); ToDeltaEndpointInjector service = new ToDeltaEndpointInjector(subject); List<Endpoint> endpoints = new LinkedList<Endpoint>(); endpoints.add(new Endpoint("localhost", 1234, "abc")); endpoints.add(new Endpoint("localhost", 2468, "xyz")); subject.onNext(endpoints); endpoints = new LinkedList<Endpoint>(); endpoints.add(new Endpoint("localhost", 1234, "abc")); endpoints.add(new Endpoint("localhost", 2468, "xyz")); subject.onNext(endpoints); endpoints = new LinkedList<Endpoint>(); endpoints.add(new Endpoint("localhost", 1234, "abc")); subject.onNext(endpoints); endpoints = new LinkedList<Endpoint>(); subject.onNext(endpoints); endpoints = new LinkedList<Endpoint>(); endpoints.add(new Endpoint("localhost", 1234, "abc")); endpoints.add(new Endpoint("localhost", 2468, "xyz")); subject.onNext(endpoints); subject.onCompleted(); BlockingObservable<EndpointChange> be = service.deltas().toBlocking(); Iterator<EndpointChange> iter = be.getIterator(); Assert.assertTrue(iter.hasNext()); EndpointChange ce2 = iter.next(); Assert.assertEquals(ce2.getEndpoint().getSlotId(), "xyz"); Assert.assertEquals(ce2.getType(), EndpointChange.Type.add); EndpointChange ce1 = iter.next(); Assert.assertEquals(ce1.getEndpoint().getSlotId(), "abc"); Assert.assertEquals(ce1.getType(), EndpointChange.Type.add); EndpointChange ce3 = iter.next(); Assert.assertEquals(ce3.getEndpoint().getSlotId(), "xyz"); Assert.assertEquals(ce3.getType(), EndpointChange.Type.complete); EndpointChange ce4 = iter.next(); Assert.assertEquals(ce4.getEndpoint().getSlotId(), "abc"); Assert.assertEquals(ce4.getType(), EndpointChange.Type.complete); EndpointChange ce6 = iter.next(); Assert.assertEquals(ce6.getEndpoint().getSlotId(), "xyz"); Assert.assertEquals(ce6.getType(), EndpointChange.Type.add); EndpointChange ce5 = iter.next(); Assert.assertEquals(ce5.getEndpoint().getSlotId(), "abc"); Assert.assertEquals(ce5.getType(), EndpointChange.Type.add); }
@Override public Observable<EndpointChange> deltas() { return Observable.merge( reconcileChanges, endpointObservable .startWith(Collections.<Endpoint>emptyList()) .buffer(2, 1) .flatMap(new Func1<List<List<Endpoint>>, Observable<EndpointChange>>() { @Override public Observable<EndpointChange> call(List<List<Endpoint>> previousAndCurrent) { if (previousAndCurrent.size() == 2) { List<Endpoint> previous = previousAndCurrent.get(0); List<Endpoint> current = previousAndCurrent.get(1); return Observable.from(changes(previous, current)); } else { return Observable.empty(); } } })); }
ToDeltaEndpointInjector implements EndpointInjector { @Override public Observable<EndpointChange> deltas() { return Observable.merge( reconcileChanges, endpointObservable .startWith(Collections.<Endpoint>emptyList()) .buffer(2, 1) .flatMap(new Func1<List<List<Endpoint>>, Observable<EndpointChange>>() { @Override public Observable<EndpointChange> call(List<List<Endpoint>> previousAndCurrent) { if (previousAndCurrent.size() == 2) { List<Endpoint> previous = previousAndCurrent.get(0); List<Endpoint> current = previousAndCurrent.get(1); return Observable.from(changes(previous, current)); } else { return Observable.empty(); } } })); } }
ToDeltaEndpointInjector implements EndpointInjector { @Override public Observable<EndpointChange> deltas() { return Observable.merge( reconcileChanges, endpointObservable .startWith(Collections.<Endpoint>emptyList()) .buffer(2, 1) .flatMap(new Func1<List<List<Endpoint>>, Observable<EndpointChange>>() { @Override public Observable<EndpointChange> call(List<List<Endpoint>> previousAndCurrent) { if (previousAndCurrent.size() == 2) { List<Endpoint> previous = previousAndCurrent.get(0); List<Endpoint> current = previousAndCurrent.get(1); return Observable.from(changes(previous, current)); } else { return Observable.empty(); } } })); } ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable);  ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable,
                                   Observable<EndpointChange> reconcileChanges); }
ToDeltaEndpointInjector implements EndpointInjector { @Override public Observable<EndpointChange> deltas() { return Observable.merge( reconcileChanges, endpointObservable .startWith(Collections.<Endpoint>emptyList()) .buffer(2, 1) .flatMap(new Func1<List<List<Endpoint>>, Observable<EndpointChange>>() { @Override public Observable<EndpointChange> call(List<List<Endpoint>> previousAndCurrent) { if (previousAndCurrent.size() == 2) { List<Endpoint> previous = previousAndCurrent.get(0); List<Endpoint> current = previousAndCurrent.get(1); return Observable.from(changes(previous, current)); } else { return Observable.empty(); } } })); } ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable);  ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable,
                                   Observable<EndpointChange> reconcileChanges); @Override Observable<EndpointChange> deltas(); }
ToDeltaEndpointInjector implements EndpointInjector { @Override public Observable<EndpointChange> deltas() { return Observable.merge( reconcileChanges, endpointObservable .startWith(Collections.<Endpoint>emptyList()) .buffer(2, 1) .flatMap(new Func1<List<List<Endpoint>>, Observable<EndpointChange>>() { @Override public Observable<EndpointChange> call(List<List<Endpoint>> previousAndCurrent) { if (previousAndCurrent.size() == 2) { List<Endpoint> previous = previousAndCurrent.get(0); List<Endpoint> current = previousAndCurrent.get(1); return Observable.from(changes(previous, current)); } else { return Observable.empty(); } } })); } ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable);  ToDeltaEndpointInjector(Observable<List<Endpoint>> endpointObservable,
                                   Observable<EndpointChange> reconcileChanges); @Override Observable<EndpointChange> deltas();  }