@Test public void setupContainerLogfileStreaming_withSomeAlreadyStreamedLogFilesAreNotIncludedNextUpdate_stopProcessingOfTheseStreams() throws Exception { dockerLogStreamManager.setupContainerLogfileStreaming( createDockerFramework(SOME_CONTAINER_ID_1, SOME_LOG_FILE_1, SOME_LOG_FILE_2)); LogStream logStreamOfFile2 = getLogStream(SOME_CONTAINER_ID_1, SOME_LOG_FILE_2); dockerLogStreamManager.setupContainerLogfileStreaming( createDockerFramework(SOME_CONTAINER_ID_1, SOME_LOG_FILE_1)); verify(streamer, times(1)).stopStreaming(eq(SOME_CONTAINER_ID_1),logStreamArgumentCaptor.capture()); assertEquals( logStreamOfFile2, logStreamArgumentCaptor.getValue()); }
public synchronized void setupContainerLogfileStreaming(DockerFramework framework) { if (!isAlreadyStreaming(framework)) { processedContainers.put(framework.getContainerId(), new HashSet<>()); } LOGGER.info("Setting up log streaming for " + framework.getName()); streamUnprocessedLogFiles(framework); stopStreamingOfOrphanLogFiles(framework); LOGGER.info("Done processing: " + framework.getName()); }
DockerLogStreamManager { public synchronized void setupContainerLogfileStreaming(DockerFramework framework) { if (!isAlreadyStreaming(framework)) { processedContainers.put(framework.getContainerId(), new HashSet<>()); } LOGGER.info("Setting up log streaming for " + framework.getName()); streamUnprocessedLogFiles(framework); stopStreamingOfOrphanLogFiles(framework); LOGGER.info("Done processing: " + framework.getName()); } }
DockerLogStreamManager { public synchronized void setupContainerLogfileStreaming(DockerFramework framework) { if (!isAlreadyStreaming(framework)) { processedContainers.put(framework.getContainerId(), new HashSet<>()); } LOGGER.info("Setting up log streaming for " + framework.getName()); streamUnprocessedLogFiles(framework); stopStreamingOfOrphanLogFiles(framework); LOGGER.info("Done processing: " + framework.getName()); } DockerLogStreamManager(DockerStreamer streamer); }
DockerLogStreamManager { public synchronized void setupContainerLogfileStreaming(DockerFramework framework) { if (!isAlreadyStreaming(framework)) { processedContainers.put(framework.getContainerId(), new HashSet<>()); } LOGGER.info("Setting up log streaming for " + framework.getName()); streamUnprocessedLogFiles(framework); stopStreamingOfOrphanLogFiles(framework); LOGGER.info("Done processing: " + framework.getName()); } DockerLogStreamManager(DockerStreamer streamer); synchronized void setupContainerLogfileStreaming(DockerFramework framework); void stopStreamingForWholeFramework(String containerId); Set<String> getProcessedContainers(); Set<DockerLogPath> getProcessedFiles(String containerId); }
DockerLogStreamManager { public synchronized void setupContainerLogfileStreaming(DockerFramework framework) { if (!isAlreadyStreaming(framework)) { processedContainers.put(framework.getContainerId(), new HashSet<>()); } LOGGER.info("Setting up log streaming for " + framework.getName()); streamUnprocessedLogFiles(framework); stopStreamingOfOrphanLogFiles(framework); LOGGER.info("Done processing: " + framework.getName()); } DockerLogStreamManager(DockerStreamer streamer); synchronized void setupContainerLogfileStreaming(DockerFramework framework); void stopStreamingForWholeFramework(String containerId); Set<String> getProcessedContainers(); Set<DockerLogPath> getProcessedFiles(String containerId);  }