ProxyRepositoryReader extends StroomZipRepositorySimpleExecutorProcessor implements Runnable { void doRunWork() { if (proxyRepositoryManager == null) { return; } final List<StroomZipRepository> readyToProcessList = proxyRepositoryManager.getReadableRepository(); for (final StroomZipRepository readyToProcess : readyToProcessList) { if (finish.get()) { return; } if (hasOutgoingRequestHandlers()) { process(readyToProcess); } readyToProcess.clean(); } } ProxyRepositoryReader();  ProxyRepositoryReader(final Monitor monitor); String getHostName(); @Override void run(); List<RequestHandler> createOutgoingRequestHandlerList(); @Override byte[] getReadBuffer(); @Override void processFeedFiles(final StroomZipRepository stroomZipRepository, final String feed,
                                 final List<Path> fileList); @StroomShutdown void stop(); boolean hasOutgoingRequestHandlers(); @StroomStartup void start(); void setForwardThreadCount(final int forwardThreadCount); void setSimpleCron(final String simpleCron); void setProxyRequestThreadLocalBuffer(final ThreadLocalBuffer proxyRequestThreadLocalBuffer); void setProxyRepositoryManager(final ProxyRepositoryManager proxyRepositoryManager); static final String PROXY_FORWARD_ID; }
@Test public void testSimpleNothingTodo() throws IOException { init(); proxyRepositoryReader.doRunWork(); }
@Test public void testSimpleOneFile() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); Assert.assertEquals(1, proxyRepository.getFileCount()); stroomZipOutputStream.close(); Assert.assertTrue(isZipFile(proxyRepository, 1)); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(1, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(2, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(0, mockRequestHandler.getHandleErrorCount()); Assert.assertEquals(2, mockRequestHandler.getEntryNameList().size()); final String sendHeader1 = new String(mockRequestHandler.getByteArray("001.meta"), StreamUtil.DEFAULT_CHARSET); Assert.assertTrue(sendHeader1.contains("Feed:TEST")); Assert.assertTrue(sendHeader1.contains("GUID:Z1")); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); }
@Test @StroomExpectedException(exception = IOException.class) public void testSimpleOneFileWithHeaderError() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "GUID:Z1\nFeed:TEST".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream.close(); mockRequestHandler.setGenerateExceptionOnHeader(true); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(0, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(0, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(1, mockRequestHandler.getHandleErrorCount()); proxyRepository.clean(); Assert.assertTrue("Expecting file to still be there", isZipFile(proxyRepository, 1)); }
@Test @StroomExpectedException(exception = IOException.class) public void testSimpleOneFileWithDataError() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream.close(); mockRequestHandler.setGenerateExceptionOnData(true); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(0, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(1, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(1, mockRequestHandler.getHandleErrorCount()); proxyRepository.clean(); Assert.assertTrue("Expecting file to still be there", isZipFile(proxyRepository, 1)); }
@Test @StroomExpectedException(exception = {IOException.class, ZipException.class}) public void testSimpleOneFileWithBadZip() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStreamImpl stroomZipOutputStream = (StroomZipOutputStreamImpl) proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream.close(); final Path zipFile = stroomZipOutputStream.getFile(); Files.delete(zipFile); OutputStream testStream = null; try { testStream = Files.newOutputStream(zipFile); testStream.write("not a zip file".getBytes(StreamUtil.DEFAULT_CHARSET)); } finally { CloseableUtil.close(testStream); } proxyRepositoryReader.doRunWork(); Assert.assertEquals(0, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(0, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(0, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(0, mockRequestHandler.getHandleErrorCount()); proxyRepository.clean(); Assert.assertTrue("Expecting bad file", isBadFile(proxyRepository, 1)); }
@Test public void testMultipleFilesWithContextSameFeed() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream1 = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry("req.header", "req", StroomZipFileType.Meta), "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry("req.data", "req", StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry("req.context", "req", StroomZipFileType.Context), "Test Context".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream1.close(); final StroomZipOutputStream stroomZipOutputStream2 = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.header", "req", StroomZipFileType.Meta), "Feed:TEST\nGUID:Z2\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.data", "req", StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.context", "req", StroomZipFileType.Context), "Test Context".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream2.close(); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(1, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(6, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(0, mockRequestHandler.getHandleErrorCount()); Assert.assertEquals(Arrays.asList("001.meta", "001.ctx", "001.dat", "002.meta", "002.ctx", "002.dat"), mockRequestHandler.getEntryNameList()); proxyRepository.clean(); final String sendHeader2 = new String(mockRequestHandler.getByteArray("002.meta"), StreamUtil.DEFAULT_CHARSET); Assert.assertTrue(sendHeader2.contains("Feed:TEST")); Assert.assertTrue(sendHeader2.contains("GUID:Z2")); Assert.assertEquals("Test Data", new String(mockRequestHandler.getByteArray("001.dat"), StreamUtil.DEFAULT_CHARSET)); Assert.assertEquals("Test Data", new String(mockRequestHandler.getByteArray("002.dat"), StreamUtil.DEFAULT_CHARSET)); Assert.assertFalse(isZipFile(proxyRepository, 1)); Assert.assertFalse(isZipFile(proxyRepository, 2)); }
@Test public void testMultipleFilesAtLimitWithContextSameFeed() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream1 = proxyRepository.getStroomZipOutputStream(); for (int i = 0; i < 10; i++) { StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Meta), "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Context), "Test Context".getBytes(StreamUtil.DEFAULT_CHARSET)); } stroomZipOutputStream1.close(); final StroomZipOutputStream stroomZipOutputStream2 = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.header", "req", StroomZipFileType.Meta), "Feed:TEST\nGUID:Z2\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.data", "req", StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream2, new StroomZipEntry("req.context", "req", StroomZipFileType.Context), "Test Context".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream2.close(); final StroomZipOutputStream stroomZipOutputStream3 = proxyRepository.getStroomZipOutputStream(); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream3, new StroomZipEntry("req.header", "req", StroomZipFileType.Meta), "Feed:TEST\nGUID:Z2\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream3, new StroomZipEntry("req.data", "req", StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream3, new StroomZipEntry("req.context", "req", StroomZipFileType.Context), "Test Context".getBytes(StreamUtil.DEFAULT_CHARSET)); stroomZipOutputStream3.close(); proxyRepositoryReader.setMaxAggregation(10); proxyRepositoryReader.doRunWork(); Assert.assertEquals(2, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(2, mockRequestHandler.getHandleFooterCount()); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); Assert.assertFalse(isZipFile(proxyRepository, 2)); }
@Test public void testMultipleFilesAtSizeLimitSameFeed() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); for (int a = 0; a < 5; a++) { final StroomZipOutputStream stroomZipOutputStream1 = proxyRepository.getStroomZipOutputStream(); for (int i = 0; i < 10; i++) { StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Meta), "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); } stroomZipOutputStream1.close(); } proxyRepositoryReader.setMaxAggregation(10000); proxyRepositoryReader.setMaxStreamSize(10L); proxyRepositoryReader.doRunWork(); Assert.assertEquals(5, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(5, mockRequestHandler.getHandleFooterCount()); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); Assert.assertFalse(isZipFile(proxyRepository, 2)); }
@Test public void testMultipleFilesAtSizeLimitSameFeed2() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); for (int a = 0; a < 5; a++) { final StroomZipOutputStream stroomZipOutputStream1 = proxyRepository.getStroomZipOutputStream(); for (int i = 0; i < 10; i++) { StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Meta), "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); } stroomZipOutputStream1.close(); } proxyRepositoryReader.setMaxAggregation(10000); proxyRepositoryReader.setMaxStreamSize(100000L); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(1, mockRequestHandler.getHandleFooterCount()); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); Assert.assertFalse(isZipFile(proxyRepository, 2)); }
@Test public void testMultipleFilesAtCountLimitSameFeed() throws IOException { init(); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); for (int a = 0; a < 5; a++) { final StroomZipOutputStream stroomZipOutputStream1 = proxyRepository.getStroomZipOutputStream(); for (int i = 0; i < 10; i++) { StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Meta), "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream1, new StroomZipEntry(null, String.valueOf(i), StroomZipFileType.Data), "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); } stroomZipOutputStream1.close(); } proxyRepositoryReader.setMaxAggregation(20); proxyRepositoryReader.setMaxStreamSize(null); proxyRepositoryReader.doRunWork(); Assert.assertEquals(3, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(3, mockRequestHandler.getHandleFooterCount()); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); Assert.assertFalse(isZipFile(proxyRepository, 2)); }
@Test public void testSimpleOneFileTemplated() throws IOException { init(); MetaMap metaMap = new MetaMap(); metaMap.put("feed", "myFeed"); metaMap.put("key1", "myKey1"); metaMap.put("key2", "myKey2"); metaMap.put("key3", "myKey3"); final String repositoryFormat = "${id}_${FEED}_${key2}_${kEy1}_${keyNotInMeta}_${Key3}"; proxyRepositoryManager.setRepositoryFormat(repositoryFormat); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream = proxyRepository.getStroomZipOutputStream(metaMap); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); Assert.assertEquals(1, proxyRepository.getFileCount()); stroomZipOutputStream.close(); final String filename = getZipFile(proxyRepository); Assert.assertEquals("001_myFeed_myKey2_myKey1___myKey3.zip", filename); Assert.assertTrue(isZipFile(proxyRepository, 1)); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(1, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(2, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(0, mockRequestHandler.getHandleErrorCount()); Assert.assertEquals(2, mockRequestHandler.getEntryNameList().size()); final String sendHeader1 = new String(mockRequestHandler.getByteArray("001.meta"), StreamUtil.DEFAULT_CHARSET); Assert.assertTrue(sendHeader1.contains("Feed:TEST")); Assert.assertTrue(sendHeader1.contains("GUID:Z1")); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); }
@Test public void testSimpleOneFileEmptyTemplate() throws IOException { init(); MetaMap metaMap = new MetaMap(); metaMap.put("feed", "myFeed"); metaMap.put("key1", "myKey1"); metaMap.put("key2", "myKey2"); metaMap.put("key3", "myKey3"); final String repositoryFormat = ""; proxyRepositoryManager.setRepositoryFormat(repositoryFormat); final StroomZipRepository proxyRepository = proxyRepositoryManager.getActiveRepository(); final StroomZipOutputStream stroomZipOutputStream = proxyRepository.getStroomZipOutputStream(metaMap); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_META_ENTRY, "Feed:TEST\nGUID:Z1\n".getBytes(StreamUtil.DEFAULT_CHARSET)); StroomZipOutputStreamUtil.addSimpleEntry(stroomZipOutputStream, StroomZipFile.SINGLE_DATA_ENTRY, "Test Data".getBytes(StreamUtil.DEFAULT_CHARSET)); Assert.assertEquals(1, proxyRepository.getFileCount()); stroomZipOutputStream.close(); final String filename = getZipFile(proxyRepository); Assert.assertEquals("001.zip", filename); Assert.assertTrue(isZipFile(proxyRepository, 1)); proxyRepositoryReader.doRunWork(); Assert.assertEquals(1, mockRequestHandler.getHandleHeaderCount()); Assert.assertEquals(1, mockRequestHandler.getHandleFooterCount()); Assert.assertEquals(2, mockRequestHandler.getHandleEntryCount()); Assert.assertEquals(0, mockRequestHandler.getHandleErrorCount()); Assert.assertEquals(2, mockRequestHandler.getEntryNameList().size()); final String sendHeader1 = new String(mockRequestHandler.getByteArray("001.meta"), StreamUtil.DEFAULT_CHARSET); Assert.assertTrue(sendHeader1.contains("Feed:TEST")); Assert.assertTrue(sendHeader1.contains("GUID:Z1")); proxyRepository.clean(); Assert.assertFalse(isZipFile(proxyRepository, 1)); }