Sender implements AutoCloseable { public void start() { log.info("Starting message sender at {}Hz", messageRate); service.scheduleAtFixedRate(this::worker, 0, 1000 / messageRate, MILLISECONDS); } @Autowired  Sender(
      AsyncRoadClient<TestMessage> onrampClient,
      TestMessageContextManager contextManager,
      @Value("${messageHz}") long messageRate);   Sender(
      AsyncRoadClient<TestMessage> onrampClient,
      TestMessageContextManager contextManager,
      long messageRate,
      ScheduledExecutorService service); void start(); void worker(); @Override void close();  }
@Test @SuppressWarnings("resource") public void _250Hz_correctly_sets_4ms_job_interval() throws Exception { new Sender(onrampClient, contextManager, 250, service).start(); verify(service).scheduleAtFixedRate(any(), eq(0L), longCaptor.capture(), timeunitCaptor.capture()); assertThat(timeunitCaptor.getValue().toMillis(longCaptor.getValue()), is(4L)); }
@Test @SuppressWarnings("unchecked") public void worker_correctly_gets_and_sends_message() throws Exception { try (Sender sender = new Sender(onrampClient, contextManager, 1L, service)) { sender.start(); verify(service).scheduleAtFixedRate(runnableCaptor.capture(), anyLong(), anyLong(), any()); TestMessageContext context = mock(TestMessageContext.class); TestMessage message = mock(TestMessage.class); CompletableFuture<StandardResponse> future = mock(CompletableFuture.class); when(context.getMessage()).thenReturn(message); when(contextManager.nextContext()).thenReturn(context); when(onrampClient.sendMessage(eq(message))).thenReturn(future); runnableCaptor.getValue().run(); verify(contextManager).nextContext(); verify(onrampClient).sendMessage(message); verify(context).messageSent(future); } }