NotificationManager extends AbstractDataSourcePropertyManager<Client, NotificationClientRepository> implements DataSink { @Override public void onNewDataComplete(DataSource source) { for (Client client : getAll(source)) { Sender<?> sender = senderCache.get(source, client); senderCache.release(source, client); sender.onNewDataComplete(); SenderResult result = sender.getSenderResult(); switch(result.getStatus()) { case SUCCESS: break; case ERROR: String errorMsg = "Failed to send notification to client " + client.getId(); if (result.getErrorCause() != null) errorMsg = errorMsg + " (" + result.getErrorCause().getMessage() + ")"; if (result.getErrorMsg() != null) errorMsg = errorMsg + " (" + result.getErrorMsg() + ")"; Log.error(errorMsg); break; case REMOVE_CLIENT: Log.info("Unregistering client " + result.getOldClient().getId()); remove(source, null, result.getOldClient()); break; case UPDATE_CLIENT: Log.info("Updating client id to " + result.getNewClient().getId()); remove(source, null, result.getOldClient()); add(source, null, result.getNewClient()); break; } } } @Inject  NotificationManager(
			Cache<NotificationClientRepository> repositoryCache,
			RepositoryFactory repositoryFactory,
			SenderCache senderCache); @Override void onNewDataStart(DataSource source); @Override void onNewDataItem(DataSource source, ObjectNode data); @Override void onNewDataComplete(DataSource source);  }
@Test public void testOnNewDataCompleteRemove() { new Expectations(notificationManager) {{ notificationManager.getAll(source); result = Arrays.asList(gcmClient); gcmSender.getSenderResult().getStatus(); result = SenderResult.Status.REMOVE_CLIENT; gcmSender.getSenderResult().getOldClient(); result = gcmClient; clientRepositoryCache.contains(anyString); result = true; clientRepositoryCache.get(anyString); result = clientRepository; }}; notificationManager.onNewDataComplete(source); new Verifications() {{ gcmSender.onNewDataComplete(); senderCache.release(source, gcmClient); clientRepository.remove(gcmClient); }}; }
@Test public void testOnNewDataCompleteUpdate() { final GcmClient newGcmClient = new GcmClient("someOtherId", "someOtherDeviceId"); new Expectations(notificationManager) {{ notificationManager.getAll(source); result = Arrays.asList(gcmClient); gcmSender.getSenderResult().getStatus(); result = SenderResult.Status.UPDATE_CLIENT; gcmSender.getSenderResult().getOldClient(); result = gcmClient; gcmSender.getSenderResult().getNewClient(); result = newGcmClient; clientRepositoryCache.contains(anyString); result = true; clientRepositoryCache.get(anyString); result = clientRepository; }}; notificationManager.onNewDataComplete(source); new Verifications() {{ gcmSender.onNewDataComplete(); senderCache.release(source, gcmClient); clientRepository.remove(gcmClient); clientRepository.add(newGcmClient); }}; }