ClusteredChangeBus extends MessageConsumer<ChangeSet> implements ChangeBus { @Override public void notify( ChangeSet changeSet ) { if (changeSet == null) { return; } if (!clusteringService.multipleMembersInCluster()) { consume(changeSet); return; } logSendOperation(changeSet); clusteringService.sendMessage(changeSet); } ClusteredChangeBus( ChangeBus delegate,
                               ClusteringService clusteringService ); @Override void consume( ChangeSet changes ); @Override synchronized void start(); @Override boolean hasObservers(); @Override synchronized void shutdown(); @Override void notify( ChangeSet changeSet ); @Override boolean register( ChangeSetListener listener ); @Override boolean registerInThread( ChangeSetListener listener ); @Override boolean unregister( ChangeSetListener listener );  }
@Test public void oneBusShouldNotifyRegisteredListeners() throws Exception { TestListener listener1 = new TestListener(1); startBusWithRegisteredListener(listener1); ChangeSet changeSet = new TestChangeSet("ws1"); ChangeBus bus1 = buses.get(1); bus1.notify(changeSet); listener1.assertExpectedEvents(changeSet); }
@Test public void twoBusesShouldNotifyEachOther() throws Exception { TestListener listener1 = new TestListener(2); TestListener listener2 = new TestListener(2); startBusWithRegisteredListener(listener1, listener2); ChangeSet changeSet1 = new TestChangeSet("bus1"); buses.get(1).notify(changeSet1); ChangeSet changeSet2 = new TestChangeSet("bus2"); buses.get(2).notify(changeSet2); listener1.assertExpectedEvents(changeSet1, changeSet2); listener2.assertExpectedEvents(changeSet1, changeSet2); }