@Test public void logAndUpdateStatus_whenMoreEventsAvailableInStore_shouldReschedule() { when(mockBackend.send(any())) .then( (Answer<BackendResponse>) invocation -> { store.persist(TRANSPORT_CONTEXT, EVENT); return BackendResponse.ok(1000); }); Iterable<PersistedEvent> persistedEvents = store.loadBatch(TRANSPORT_CONTEXT); uploader.logAndUpdateState(TRANSPORT_CONTEXT, 1); verify(mockScheduler, times(1)).schedule(TRANSPORT_CONTEXT, 1); }
void logAndUpdateState(TransportContext transportContext, int attemptNumber) { TransportBackend backend = backendRegistry.get(transportContext.getBackendName()); Iterable<PersistedEvent> persistedEvents = guard.runCriticalSection(() -> eventStore.loadBatch(transportContext)); if (!persistedEvents.iterator().hasNext()) { return; } BackendResponse response; if (backend == null) { Logging.d( LOG_TAG, "Unknown backend for %s, deleting event batch for it...", transportContext); response = BackendResponse.fatalError(); } else { List<EventInternal> eventInternals = new ArrayList<>(); for (PersistedEvent persistedEvent : persistedEvents) { eventInternals.add(persistedEvent.getEvent()); } response = backend.send( BackendRequest.builder() .setEvents(eventInternals) .setExtras(transportContext.getExtras()) .build()); } guard.runCriticalSection( () -> { if (response.getStatus() == BackendResponse.Status.TRANSIENT_ERROR) { eventStore.recordFailure(persistedEvents); workScheduler.schedule(transportContext, attemptNumber + 1); } else { eventStore.recordSuccess(persistedEvents); if (response.getStatus() == BackendResponse.Status.OK) { eventStore.recordNextCallTime( transportContext, clock.getTime() + response.getNextRequestWaitMillis()); } if (eventStore.hasPendingEventsFor(transportContext)) { workScheduler.schedule(transportContext, 1); } } return null; }); }
Uploader { void logAndUpdateState(TransportContext transportContext, int attemptNumber) { TransportBackend backend = backendRegistry.get(transportContext.getBackendName()); Iterable<PersistedEvent> persistedEvents = guard.runCriticalSection(() -> eventStore.loadBatch(transportContext)); if (!persistedEvents.iterator().hasNext()) { return; } BackendResponse response; if (backend == null) { Logging.d( LOG_TAG, "Unknown backend for %s, deleting event batch for it...", transportContext); response = BackendResponse.fatalError(); } else { List<EventInternal> eventInternals = new ArrayList<>(); for (PersistedEvent persistedEvent : persistedEvents) { eventInternals.add(persistedEvent.getEvent()); } response = backend.send( BackendRequest.builder() .setEvents(eventInternals) .setExtras(transportContext.getExtras()) .build()); } guard.runCriticalSection( () -> { if (response.getStatus() == BackendResponse.Status.TRANSIENT_ERROR) { eventStore.recordFailure(persistedEvents); workScheduler.schedule(transportContext, attemptNumber + 1); } else { eventStore.recordSuccess(persistedEvents); if (response.getStatus() == BackendResponse.Status.OK) { eventStore.recordNextCallTime( transportContext, clock.getTime() + response.getNextRequestWaitMillis()); } if (eventStore.hasPendingEventsFor(transportContext)) { workScheduler.schedule(transportContext, 1); } } return null; }); } }
Uploader { void logAndUpdateState(TransportContext transportContext, int attemptNumber) { TransportBackend backend = backendRegistry.get(transportContext.getBackendName()); Iterable<PersistedEvent> persistedEvents = guard.runCriticalSection(() -> eventStore.loadBatch(transportContext)); if (!persistedEvents.iterator().hasNext()) { return; } BackendResponse response; if (backend == null) { Logging.d( LOG_TAG, "Unknown backend for %s, deleting event batch for it...", transportContext); response = BackendResponse.fatalError(); } else { List<EventInternal> eventInternals = new ArrayList<>(); for (PersistedEvent persistedEvent : persistedEvents) { eventInternals.add(persistedEvent.getEvent()); } response = backend.send( BackendRequest.builder() .setEvents(eventInternals) .setExtras(transportContext.getExtras()) .build()); } guard.runCriticalSection( () -> { if (response.getStatus() == BackendResponse.Status.TRANSIENT_ERROR) { eventStore.recordFailure(persistedEvents); workScheduler.schedule(transportContext, attemptNumber + 1); } else { eventStore.recordSuccess(persistedEvents); if (response.getStatus() == BackendResponse.Status.OK) { eventStore.recordNextCallTime( transportContext, clock.getTime() + response.getNextRequestWaitMillis()); } if (eventStore.hasPendingEventsFor(transportContext)) { workScheduler.schedule(transportContext, 1); } } return null; }); } @Inject  Uploader(
      Context context,
      BackendRegistry backendRegistry,
      EventStore eventStore,
      WorkScheduler workScheduler,
      Executor executor,
      SynchronizationGuard guard,
      @WallTime Clock clock); }
Uploader { void logAndUpdateState(TransportContext transportContext, int attemptNumber) { TransportBackend backend = backendRegistry.get(transportContext.getBackendName()); Iterable<PersistedEvent> persistedEvents = guard.runCriticalSection(() -> eventStore.loadBatch(transportContext)); if (!persistedEvents.iterator().hasNext()) { return; } BackendResponse response; if (backend == null) { Logging.d( LOG_TAG, "Unknown backend for %s, deleting event batch for it...", transportContext); response = BackendResponse.fatalError(); } else { List<EventInternal> eventInternals = new ArrayList<>(); for (PersistedEvent persistedEvent : persistedEvents) { eventInternals.add(persistedEvent.getEvent()); } response = backend.send( BackendRequest.builder() .setEvents(eventInternals) .setExtras(transportContext.getExtras()) .build()); } guard.runCriticalSection( () -> { if (response.getStatus() == BackendResponse.Status.TRANSIENT_ERROR) { eventStore.recordFailure(persistedEvents); workScheduler.schedule(transportContext, attemptNumber + 1); } else { eventStore.recordSuccess(persistedEvents); if (response.getStatus() == BackendResponse.Status.OK) { eventStore.recordNextCallTime( transportContext, clock.getTime() + response.getNextRequestWaitMillis()); } if (eventStore.hasPendingEventsFor(transportContext)) { workScheduler.schedule(transportContext, 1); } } return null; }); } @Inject  Uploader(
      Context context,
      BackendRegistry backendRegistry,
      EventStore eventStore,
      WorkScheduler workScheduler,
      Executor executor,
      SynchronizationGuard guard,
      @WallTime Clock clock); void upload(TransportContext transportContext, int attemptNumber, Runnable callback); }
Uploader { void logAndUpdateState(TransportContext transportContext, int attemptNumber) { TransportBackend backend = backendRegistry.get(transportContext.getBackendName()); Iterable<PersistedEvent> persistedEvents = guard.runCriticalSection(() -> eventStore.loadBatch(transportContext)); if (!persistedEvents.iterator().hasNext()) { return; } BackendResponse response; if (backend == null) { Logging.d( LOG_TAG, "Unknown backend for %s, deleting event batch for it...", transportContext); response = BackendResponse.fatalError(); } else { List<EventInternal> eventInternals = new ArrayList<>(); for (PersistedEvent persistedEvent : persistedEvents) { eventInternals.add(persistedEvent.getEvent()); } response = backend.send( BackendRequest.builder() .setEvents(eventInternals) .setExtras(transportContext.getExtras()) .build()); } guard.runCriticalSection( () -> { if (response.getStatus() == BackendResponse.Status.TRANSIENT_ERROR) { eventStore.recordFailure(persistedEvents); workScheduler.schedule(transportContext, attemptNumber + 1); } else { eventStore.recordSuccess(persistedEvents); if (response.getStatus() == BackendResponse.Status.OK) { eventStore.recordNextCallTime( transportContext, clock.getTime() + response.getNextRequestWaitMillis()); } if (eventStore.hasPendingEventsFor(transportContext)) { workScheduler.schedule(transportContext, 1); } } return null; }); } @Inject  Uploader(
      Context context,
      BackendRegistry backendRegistry,
      EventStore eventStore,
      WorkScheduler workScheduler,
      Executor executor,
      SynchronizationGuard guard,
      @WallTime Clock clock); void upload(TransportContext transportContext, int attemptNumber, Runnable callback);  }