@Test public void testBackwardCompatible() throws IOException { try (final InputStream fis = new FileInputStream(new File("src/test/resources/old-swap-file.swap")); final DataInputStream in = new DataInputStream(new BufferedInputStream(fis))) { final FlowFileQueue flowFileQueue = Mockito.mock(FlowFileQueue.class); Mockito.when(flowFileQueue.getIdentifier()).thenReturn("87bb99fe-412c-49f6-a441-d1b0af4e20b4"); final FileSystemSwapManager swapManager = createSwapManager(); final SwapContents swapContents = swapManager.peek("src/test/resources/old-swap-file.swap", flowFileQueue); final List<FlowFileRecord> records = swapContents.getFlowFiles(); assertEquals(10000, records.size()); for (final FlowFileRecord record : records) { assertEquals(4, record.getAttributes().size()); assertEquals("value", record.getAttribute("key")); } } }
@Override public SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); if (!swapFile.exists()) { throw new FileNotFoundException("Failed to swap in FlowFiles from external storage location " + swapLocation + " into FlowFile Queue because the file could not be found"); } try (final InputStream fis = new FileInputStream(swapFile); final InputStream bis = new BufferedInputStream(fis); final DataInputStream in = new DataInputStream(bis)) { final SwapDeserializer deserializer = createSwapDeserializer(in); return deserializer.deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } }
FileSystemSwapManager implements FlowFileSwapManager { @Override public SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); if (!swapFile.exists()) { throw new FileNotFoundException("Failed to swap in FlowFiles from external storage location " + swapLocation + " into FlowFile Queue because the file could not be found"); } try (final InputStream fis = new FileInputStream(swapFile); final InputStream bis = new BufferedInputStream(fis); final DataInputStream in = new DataInputStream(bis)) { final SwapDeserializer deserializer = createSwapDeserializer(in); return deserializer.deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } } }
FileSystemSwapManager implements FlowFileSwapManager { @Override public SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); if (!swapFile.exists()) { throw new FileNotFoundException("Failed to swap in FlowFiles from external storage location " + swapLocation + " into FlowFile Queue because the file could not be found"); } try (final InputStream fis = new FileInputStream(swapFile); final InputStream bis = new BufferedInputStream(fis); final DataInputStream in = new DataInputStream(bis)) { final SwapDeserializer deserializer = createSwapDeserializer(in); return deserializer.deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } } FileSystemSwapManager();  FileSystemSwapManager(final NiFiProperties nifiProperties); }
FileSystemSwapManager implements FlowFileSwapManager { @Override public SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); if (!swapFile.exists()) { throw new FileNotFoundException("Failed to swap in FlowFiles from external storage location " + swapLocation + " into FlowFile Queue because the file could not be found"); } try (final InputStream fis = new FileInputStream(swapFile); final InputStream bis = new BufferedInputStream(fis); final DataInputStream in = new DataInputStream(bis)) { final SwapDeserializer deserializer = createSwapDeserializer(in); return deserializer.deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } } FileSystemSwapManager();  FileSystemSwapManager(final NiFiProperties nifiProperties); @Override synchronized void initialize(final SwapManagerInitializationContext initializationContext); @Override String swapOut(final List<FlowFileRecord> toSwap, final FlowFileQueue flowFileQueue); @Override SwapContents swapIn(final String swapLocation, final FlowFileQueue flowFileQueue); @Override SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue); @Override void purge(); @Override List<String> recoverSwapLocations(final FlowFileQueue flowFileQueue); @Override SwapSummary getSwapSummary(final String swapLocation); }
FileSystemSwapManager implements FlowFileSwapManager { @Override public SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); if (!swapFile.exists()) { throw new FileNotFoundException("Failed to swap in FlowFiles from external storage location " + swapLocation + " into FlowFile Queue because the file could not be found"); } try (final InputStream fis = new FileInputStream(swapFile); final InputStream bis = new BufferedInputStream(fis); final DataInputStream in = new DataInputStream(bis)) { final SwapDeserializer deserializer = createSwapDeserializer(in); return deserializer.deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } } FileSystemSwapManager();  FileSystemSwapManager(final NiFiProperties nifiProperties); @Override synchronized void initialize(final SwapManagerInitializationContext initializationContext); @Override String swapOut(final List<FlowFileRecord> toSwap, final FlowFileQueue flowFileQueue); @Override SwapContents swapIn(final String swapLocation, final FlowFileQueue flowFileQueue); @Override SwapContents peek(final String swapLocation, final FlowFileQueue flowFileQueue); @Override void purge(); @Override List<String> recoverSwapLocations(final FlowFileQueue flowFileQueue); @Override SwapSummary getSwapSummary(final String swapLocation); static final int MINIMUM_SWAP_COUNT; static final int SWAP_ENCODING_VERSION; static final String EVENT_CATEGORY; }