@Ignore("test is brittle as depends on timing - not reliable across systems, in parallel builds, etc..") @Test public void testPartialInteraction() throws Exception { int port = this.availablePort(); Server server = setupTCPServer(port); testRunner.setProperty(GetTCP.ENDPOINT_LIST, "localhost:" + port); testRunner.setProperty(GetTCP.RECEIVE_BUFFER_SIZE, "2B"); testRunner.run(1000, false); this.sendToSocket(new InetSocketAddress(port), "Hello\r"); Thread.sleep(300); testRunner.assertAllFlowFilesTransferred(GetTCP.REL_PARTIAL, 3); testRunner.clearTransferState(); this.sendToSocket(new InetSocketAddress(port), "H\r"); Thread.sleep(300); testRunner.assertAllFlowFilesTransferred(GetTCP.REL_SUCCESS, 1); testRunner.clearTransferState(); testRunner.shutdown(); server.stop(); }
private void run(ProcessContext context) { String[] serverAddresses = this.originalServerAddressList.split(","); for (String hostPortPair : serverAddresses) { ReceivingClient client; if (!this.liveTcpClients.containsKey(hostPortPair)) { String[] hostAndPort = hostPortPair.split(":"); InetSocketAddress address = new InetSocketAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1])); client = new ReceivingClient(address, this.clientScheduler, this.receiveBufferSize, this.endOfMessageByte); client.setReconnectAttempts(this.connectionAttemptCount); client.setDelayMillisBeforeReconnect(this.reconnectInterval); client.setMessageHandler(this.delegatingMessageHandler); this.liveTcpClients.put(hostPortPair, client); this.startClient(client); } else { client = this.liveTcpClients.get(hostPortPair); if (!client.isRunning()) { client.stop(); this.startClient(client); } } } }
GetTCP extends AbstractSessionFactoryProcessor { private void run(ProcessContext context) { String[] serverAddresses = this.originalServerAddressList.split(","); for (String hostPortPair : serverAddresses) { ReceivingClient client; if (!this.liveTcpClients.containsKey(hostPortPair)) { String[] hostAndPort = hostPortPair.split(":"); InetSocketAddress address = new InetSocketAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1])); client = new ReceivingClient(address, this.clientScheduler, this.receiveBufferSize, this.endOfMessageByte); client.setReconnectAttempts(this.connectionAttemptCount); client.setDelayMillisBeforeReconnect(this.reconnectInterval); client.setMessageHandler(this.delegatingMessageHandler); this.liveTcpClients.put(hostPortPair, client); this.startClient(client); } else { client = this.liveTcpClients.get(hostPortPair); if (!client.isRunning()) { client.stop(); this.startClient(client); } } } } }
GetTCP extends AbstractSessionFactoryProcessor { private void run(ProcessContext context) { String[] serverAddresses = this.originalServerAddressList.split(","); for (String hostPortPair : serverAddresses) { ReceivingClient client; if (!this.liveTcpClients.containsKey(hostPortPair)) { String[] hostAndPort = hostPortPair.split(":"); InetSocketAddress address = new InetSocketAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1])); client = new ReceivingClient(address, this.clientScheduler, this.receiveBufferSize, this.endOfMessageByte); client.setReconnectAttempts(this.connectionAttemptCount); client.setDelayMillisBeforeReconnect(this.reconnectInterval); client.setMessageHandler(this.delegatingMessageHandler); this.liveTcpClients.put(hostPortPair, client); this.startClient(client); } else { client = this.liveTcpClients.get(hostPortPair); if (!client.isRunning()) { client.stop(); this.startClient(client); } } } }  }
GetTCP extends AbstractSessionFactoryProcessor { private void run(ProcessContext context) { String[] serverAddresses = this.originalServerAddressList.split(","); for (String hostPortPair : serverAddresses) { ReceivingClient client; if (!this.liveTcpClients.containsKey(hostPortPair)) { String[] hostAndPort = hostPortPair.split(":"); InetSocketAddress address = new InetSocketAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1])); client = new ReceivingClient(address, this.clientScheduler, this.receiveBufferSize, this.endOfMessageByte); client.setReconnectAttempts(this.connectionAttemptCount); client.setDelayMillisBeforeReconnect(this.reconnectInterval); client.setMessageHandler(this.delegatingMessageHandler); this.liveTcpClients.put(hostPortPair, client); this.startClient(client); } else { client = this.liveTcpClients.get(hostPortPair); if (!client.isRunning()) { client.stop(); this.startClient(client); } } } }  @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @Override void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory); @OnStopped void tearDown(); }
GetTCP extends AbstractSessionFactoryProcessor { private void run(ProcessContext context) { String[] serverAddresses = this.originalServerAddressList.split(","); for (String hostPortPair : serverAddresses) { ReceivingClient client; if (!this.liveTcpClients.containsKey(hostPortPair)) { String[] hostAndPort = hostPortPair.split(":"); InetSocketAddress address = new InetSocketAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1])); client = new ReceivingClient(address, this.clientScheduler, this.receiveBufferSize, this.endOfMessageByte); client.setReconnectAttempts(this.connectionAttemptCount); client.setDelayMillisBeforeReconnect(this.reconnectInterval); client.setMessageHandler(this.delegatingMessageHandler); this.liveTcpClients.put(hostPortPair, client); this.startClient(client); } else { client = this.liveTcpClients.get(hostPortPair); if (!client.isRunning()) { client.stop(); this.startClient(client); } } } }  @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @Override void onTrigger(ProcessContext context, ProcessSessionFactory sessionFactory); @OnStopped void tearDown(); static final PropertyDescriptor ENDPOINT_LIST; static final PropertyDescriptor CONNECTION_ATTEMPT_COUNT; static final PropertyDescriptor RECONNECT_INTERVAL; static final PropertyDescriptor RECEIVE_BUFFER_SIZE; static final PropertyDescriptor END_OF_MESSAGE_BYTE; static final Relationship REL_SUCCESS; static final Relationship REL_PARTIAL; }