@Test public void testResponder() throws Exception { String id = UUID.randomUUID().toString(); StandardEvent event = new StandardEvent("Test", null, new MockChannelResponder()); service.register(id, new TCPContextEvent(null, Arrays.asList(event))); final Map<String, String> attributes = new HashMap<>(); attributes.put(ListenTCP.TCP_SENDER, "Test"); attributes.put(ListenTCP.TCP_PORT, "8800"); attributes.put(ListenTCP.TCP_CONTEXT_ID, id); attributes.put(ListenTCP.TCP_CONTEXT_CHARSET, StandardCharsets.UTF_8.name()); runner.enqueue("hello".getBytes(), attributes); runner.run(); final List<MockFlowFile> outputFlowFiles = runner.getFlowFilesForRelationship(HandleTCPResponse.REL_SUCCESS); final List<MockFlowFile> originalFlowFiles = runner.getFlowFilesForRelationship(HandleTCPResponse.REL_ORIGINAL); assertEquals(1, outputFlowFiles.size()); assertEquals(1, originalFlowFiles.size()); assertTrue(runner.getProvenanceEvents().size() == 1); assertEquals(ProvenanceEventType.SEND, runner.getProvenanceEvents().get(0).getEventType()); assertEquals("tcp: }
protected String getTransitUri(Map<String, String> attributes) { final String sender = attributes.get(ListenTCP.TCP_SENDER); final String port = attributes.get(ListenTCP.TCP_PORT); return ListenTCP.getTransitUri(sender, port); }
HandleTCPResponse extends AbstractProcessor { protected String getTransitUri(Map<String, String> attributes) { final String sender = attributes.get(ListenTCP.TCP_SENDER); final String port = attributes.get(ListenTCP.TCP_PORT); return ListenTCP.getTransitUri(sender, port); } }
HandleTCPResponse extends AbstractProcessor { protected String getTransitUri(Map<String, String> attributes) { final String sender = attributes.get(ListenTCP.TCP_SENDER); final String port = attributes.get(ListenTCP.TCP_PORT); return ListenTCP.getTransitUri(sender, port); }  }
HandleTCPResponse extends AbstractProcessor { protected String getTransitUri(Map<String, String> attributes) { final String sender = attributes.get(ListenTCP.TCP_SENDER); final String port = attributes.get(ListenTCP.TCP_PORT); return ListenTCP.getTransitUri(sender, port); }  @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override Set<Relationship> getRelationships(); @OnScheduled void onScheduled(ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
HandleTCPResponse extends AbstractProcessor { protected String getTransitUri(Map<String, String> attributes) { final String sender = attributes.get(ListenTCP.TCP_SENDER); final String port = attributes.get(ListenTCP.TCP_PORT); return ListenTCP.getTransitUri(sender, port); }  @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @Override Set<Relationship> getRelationships(); @OnScheduled void onScheduled(ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); static final String TCP_RESPONSE_SENDER; static final PropertyDescriptor RESPONDER_CONTEXT_MAP; static final PropertyDescriptor RESPONSE_DELIMITER; static final PropertyDescriptor RESPONSE_TYPE; static final PropertyDescriptor RESPONSE_MAX_DATA_LENGTH; static final PropertyDescriptor RESPONSE_TEXT; static final Relationship REL_SUCCESS; static final Relationship REL_ORIGINAL; static final Relationship REL_FAILURE; }