@Test public void setupContainerLogfileStreaming_withSeveralNewLogFiles_shouldStreamAllOfThem() throws Exception { DockerFramework framework = createDockerFramework(SOME_CONTAINER_ID_1, SOME_LOG_FILE_1, SOME_LOG_FILE_2); dockerLogStreamManager.setupContainerLogfileStreaming(framework); verify(streamer, times(2)).startStreaming(dockerLogPathArgumentCaptor.capture()); assertEquals( new DockerLogPath(SOME_CONTAINER_ID_1, SOME_FRAMEWORK_NAME, SOME_LOG_FILE_1), dockerLogPathArgumentCaptor.getAllValues().get(0)); assertEquals( new DockerLogPath(SOME_CONTAINER_ID_1, SOME_FRAMEWORK_NAME, SOME_LOG_FILE_2), dockerLogPathArgumentCaptor.getAllValues().get(1)); }
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);  }