@Test public void testUpdate() throws Exception { FunctionMetaData request = FunctionMetaData.newBuilder().build(); Message msg = mock(Message.class); when(msg.getData()).thenReturn(request.toByteArray()); CountDownLatch readLatch = new CountDownLatch(1); CountDownLatch processLatch = new CountDownLatch(1); when(reader.readNext(anyInt(), any(TimeUnit.class))).thenReturn(msg).then(new Answer<Message>() { public Message answer(InvocationOnMock invocation) { try { readLatch.countDown(); processLatch.await(); } catch (Exception e) { throw new RuntimeException(e); } return null; } }); fsc.start(); readLatch.await(); verify(reader, times(2)).readNext(anyInt(), any(TimeUnit.class)); verify(fsm, times(1)).processMetaDataTopicMessage(any(Message.class)); }
public void start() { isRunning = true; tailerThread.start(); }
FunctionMetaDataTopicTailer implements Runnable, AutoCloseable { public void start() { isRunning = true; tailerThread.start(); } }
FunctionMetaDataTopicTailer implements Runnable, AutoCloseable { public void start() { isRunning = true; tailerThread.start(); } FunctionMetaDataTopicTailer(FunctionMetaDataManager functionMetaDataManager,
                                       ReaderBuilder readerBuilder, WorkerConfig workerConfig,
                                       MessageId lastMessageSeen,
                                       ErrorNotifier errorNotifier); }
FunctionMetaDataTopicTailer implements Runnable, AutoCloseable { public void start() { isRunning = true; tailerThread.start(); } FunctionMetaDataTopicTailer(FunctionMetaDataManager functionMetaDataManager,
                                       ReaderBuilder readerBuilder, WorkerConfig workerConfig,
                                       MessageId lastMessageSeen,
                                       ErrorNotifier errorNotifier); void start(); @Override void run(); CompletableFuture<Void> stopWhenNoMoreMessages(); @Override void close(); static Reader createReader(WorkerConfig workerConfig, ReaderBuilder readerBuilder,
                                      MessageId startMessageId); }
FunctionMetaDataTopicTailer implements Runnable, AutoCloseable { public void start() { isRunning = true; tailerThread.start(); } FunctionMetaDataTopicTailer(FunctionMetaDataManager functionMetaDataManager,
                                       ReaderBuilder readerBuilder, WorkerConfig workerConfig,
                                       MessageId lastMessageSeen,
                                       ErrorNotifier errorNotifier); void start(); @Override void run(); CompletableFuture<Void> stopWhenNoMoreMessages(); @Override void close(); static Reader createReader(WorkerConfig workerConfig, ReaderBuilder readerBuilder,
                                      MessageId startMessageId);  }