NiFiSinglePortOutputOperator extends AbstractNiFiOutputOperator<T> { @Override protected void endNewWindow() { inputPort.flush(); } private  NiFiSinglePortOutputOperator();  NiFiSinglePortOutputOperator(
      final SiteToSiteClient.Builder siteToSiteBuilder,
      final NiFiDataPacketBuilder<T> dataPacketBuilder,
      final WindowDataManager windowDataManager,
      final int batchSize);  final transient BufferingInputPort inputPort; }
@Test public void testTransactionPerTuple() throws IOException { operator.setup(context); final MockSiteToSiteClient mockClient = (MockSiteToSiteClient)operator.client; final String tuple1 = "tuple1"; final String tuple2 = "tuple2"; final String tuple3 = "tuple3"; operator.beginWindow(1); operator.inputPort.process(tuple1); Assert.assertEquals(1, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple2); Assert.assertEquals(2, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple3); Assert.assertEquals(3, mockClient.getMockTransactions().size()); operator.endNewWindow(); Assert.assertEquals(3, mockClient.getMockTransactions().size()); List<String> expectedContents = Arrays.asList(tuple1, tuple2, tuple3); List<MockTransaction> transactions = mockClient.getMockTransactions(); verifyTransactions(expectedContents, transactions); }
@Test public void testBatchSize() throws IOException { final int batchSize = 3; operator = new NiFiSinglePortOutputOperator(stsBuilder, dpBuilder, windowDataManager, batchSize); operator.setup(context); final MockSiteToSiteClient mockClient = (MockSiteToSiteClient)operator.client; final String tuple1 = "tuple1"; final String tuple2 = "tuple2"; final String tuple3 = "tuple3"; final String tuple4 = "tuple4"; final String tuple5 = "tuple5"; operator.beginWindow(1); operator.inputPort.process(tuple1); Assert.assertEquals(0, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple2); Assert.assertEquals(0, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple3); Assert.assertEquals(1, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple4); Assert.assertEquals(1, mockClient.getMockTransactions().size()); operator.inputPort.process(tuple5); Assert.assertEquals(1, mockClient.getMockTransactions().size()); operator.endNewWindow(); Assert.assertEquals(2, mockClient.getMockTransactions().size()); List<String> expectedContents = Arrays.asList(tuple1, tuple2, tuple3, tuple4, tuple5); List<MockTransaction> transactions = mockClient.getMockTransactions(); verifyTransactions(expectedContents, transactions); }