http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java index 3f0ec4f..e7b6d06 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/DataPacket.java @@ -19,27 +19,30 @@ package org.apache.nifi.remote.protocol; import java.io.InputStream; import java.util.Map; - /** - * Represents a piece of data that is to be sent to or that was received from a NiFi instance. + * Represents a piece of data that is to be sent to or that was received from a + * NiFi instance. */ public interface DataPacket { /** * The key-value attributes that are to be associated with the data - * @return + * + * @return all attributes + */ + Map<String, String> getAttributes(); + + /** + * An InputStream from which the content can be read + * + * @return input stream to the data */ - Map<String, String> getAttributes(); - - /** - * An InputStream from which the content can be read - * @return - */ - InputStream getData(); + InputStream getData(); - /** - * The length of the InputStream. - * @return - */ - long getSize(); + /** + * The length of the InputStream. + * + * @return length of the inputstream. + */ + long getSize(); }
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java index 41dc276..016690c 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/HandshakeProperty.java @@ -16,46 +16,44 @@ */ package org.apache.nifi.remote.protocol.socket; - /** - * Enumeration of Properties that can be used for the Site-to-Site Socket Protocol. + * Enumeration of Properties that can be used for the Site-to-Site Socket + * Protocol. */ public enum HandshakeProperty { + /** - * Boolean value indicating whether or not the contents of a FlowFile should be - * GZipped when transferred. + * Boolean value indicating whether or not the contents of a FlowFile should + * be GZipped when transferred. */ GZIP, - /** * The unique identifier of the port to communicate with */ PORT_IDENTIFIER, - /** - * Indicates the number of milliseconds after the request was made that the client - * will wait for a response. If no response has been received by the time this value - * expires, the server can move on without attempting to service the request because - * the client will have already disconnected. + * Indicates the number of milliseconds after the request was made that the + * client will wait for a response. If no response has been received by the + * time this value expires, the server can move on without attempting to + * service the request because the client will have already disconnected. */ REQUEST_EXPIRATION_MILLIS, - /** - * The preferred number of FlowFiles that the server should send to the client - * when pulling data. This property was introduced in version 5 of the protocol. + * The preferred number of FlowFiles that the server should send to the + * client when pulling data. This property was introduced in version 5 of + * the protocol. */ BATCH_COUNT, - /** - * The preferred number of bytes that the server should send to the client when - * pulling data. This property was introduced in version 5 of the protocol. + * The preferred number of bytes that the server should send to the client + * when pulling data. This property was introduced in version 5 of the + * protocol. */ BATCH_SIZE, - /** - * The preferred amount of time that the server should send data to the client - * when pulling data. This property was introduced in version 5 of the protocol. - * Value is in milliseconds. + * The preferred amount of time that the server should send data to the + * client when pulling data. This property was introduced in version 5 of + * the protocol. Value is in milliseconds. */ BATCH_DURATION; } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java index eae1940..6ad2ba0 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/Response.java @@ -22,28 +22,29 @@ import java.io.IOException; import org.apache.nifi.remote.exception.ProtocolException; public class Response { + private final ResponseCode code; private final String message; - + private Response(final ResponseCode code, final String explanation) { this.code = code; this.message = explanation; } - + public ResponseCode getCode() { return code; } - + public String getMessage() { return message; } - + public static Response read(final DataInputStream in) throws IOException, ProtocolException { final ResponseCode code = ResponseCode.readCode(in); final String message = code.containsMessage() ? in.readUTF() : null; return new Response(code, message); } - + @Override public String toString() { return code + ": " + message; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java index 8860e73..0e1359e 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/ResponseCode.java @@ -23,131 +23,126 @@ import java.io.InputStream; import org.apache.nifi.remote.exception.ProtocolException; - public enum ResponseCode { + RESERVED(0, "Reserved for Future Use", false), // This will likely be used if we ever need to expand the length of - // ResponseCode, so that we can indicate a 0 followed by some other bytes - + // ResponseCode, so that we can indicate a 0 followed by some other bytes + // handshaking properties PROPERTIES_OK(1, "Properties OK", false), UNKNOWN_PROPERTY_NAME(230, "Unknown Property Name", true), ILLEGAL_PROPERTY_VALUE(231, "Illegal Property Value", true), MISSING_PROPERTY(232, "Missing Property", true), - // transaction indicators CONTINUE_TRANSACTION(10, "Continue Transaction", false), FINISH_TRANSACTION(11, "Finish Transaction", false), - CONFIRM_TRANSACTION(12, "Confirm Transaction", true), // "Explanation" of this code is the checksum + CONFIRM_TRANSACTION(12, "Confirm Transaction", true), // "Explanation" of this code is the checksum TRANSACTION_FINISHED(13, "Transaction Finished", false), TRANSACTION_FINISHED_BUT_DESTINATION_FULL(14, "Transaction Finished But Destination is Full", false), CANCEL_TRANSACTION(15, "Cancel Transaction", true), BAD_CHECKSUM(19, "Bad Checksum", false), - // data availability indicators MORE_DATA(20, "More Data Exists", false), NO_MORE_DATA(21, "No More Data Exists", false), - // port state indicators UNKNOWN_PORT(200, "Unknown Port", false), PORT_NOT_IN_VALID_STATE(201, "Port Not in a Valid State", true), PORTS_DESTINATION_FULL(202, "Port's Destination is Full", false), - // authorization UNAUTHORIZED(240, "User Not Authorized", true), - // error indicators ABORT(250, "Abort", true), UNRECOGNIZED_RESPONSE_CODE(254, "Unrecognized Response Code", false), END_OF_STREAM(255, "End of Stream", false); - + private static final ResponseCode[] codeArray = new ResponseCode[256]; - + static { - for ( final ResponseCode responseCode : ResponseCode.values() ) { + for (final ResponseCode responseCode : ResponseCode.values()) { codeArray[responseCode.getCode()] = responseCode; } } - + private static final byte CODE_SEQUENCE_VALUE_1 = (byte) 'R'; private static final byte CODE_SEQUENCE_VALUE_2 = (byte) 'C'; private final int code; private final byte[] codeSequence; private final String description; private final boolean containsMessage; - + private ResponseCode(final int code, final String description, final boolean containsMessage) { - this.codeSequence = new byte[] {CODE_SEQUENCE_VALUE_1, CODE_SEQUENCE_VALUE_2, (byte) code}; + this.codeSequence = new byte[]{CODE_SEQUENCE_VALUE_1, CODE_SEQUENCE_VALUE_2, (byte) code}; this.code = code; this.description = description; this.containsMessage = containsMessage; } - + public int getCode() { return code; } - + public byte[] getCodeSequence() { return codeSequence; } - + @Override public String toString() { return description; } - + public boolean containsMessage() { return containsMessage; } - + public void writeResponse(final DataOutputStream out) throws IOException { - if ( containsMessage() ) { + if (containsMessage()) { throw new IllegalArgumentException("ResponseCode " + code + " expects an explanation"); } - + out.write(getCodeSequence()); out.flush(); } - + public void writeResponse(final DataOutputStream out, final String explanation) throws IOException { - if ( !containsMessage() ) { + if (!containsMessage()) { throw new IllegalArgumentException("ResponseCode " + code + " does not expect an explanation"); } - + out.write(getCodeSequence()); out.writeUTF(explanation); out.flush(); } - + static ResponseCode readCode(final InputStream in) throws IOException, ProtocolException { final int byte1 = in.read(); - if ( byte1 < 0 ) { + if (byte1 < 0) { throw new EOFException(); - } else if ( byte1 != CODE_SEQUENCE_VALUE_1 ) { + } else if (byte1 != CODE_SEQUENCE_VALUE_1) { throw new ProtocolException("Expected to receive ResponseCode, but the stream did not have a ResponseCode"); } - + final int byte2 = in.read(); - if ( byte2 < 0 ) { + if (byte2 < 0) { throw new EOFException(); - } else if ( byte2 != CODE_SEQUENCE_VALUE_2 ) { + } else if (byte2 != CODE_SEQUENCE_VALUE_2) { throw new ProtocolException("Expected to receive ResponseCode, but the stream did not have a ResponseCode"); } final int byte3 = in.read(); - if ( byte3 < 0 ) { + if (byte3 < 0) { throw new EOFException(); } - + final ResponseCode responseCode = codeArray[byte3]; if (responseCode == null) { throw new ProtocolException("Received Response Code of " + byte3 + " but do not recognize this code"); } return responseCode; } - + public static ResponseCode fromSequence(final byte[] value) { final int code = value[3] & 0xFF; final ResponseCode responseCode = codeArray[code]; return (responseCode == null) ? UNRECOGNIZED_RESPONSE_CODE : responseCode; } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java index 83c5305..de845ee 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientProtocol.java @@ -58,120 +58,121 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class SocketClientProtocol implements ClientProtocol { + private final VersionNegotiator versionNegotiator = new StandardVersionNegotiator(5, 4, 3, 2, 1); private RemoteDestination destination; private boolean useCompression = false; - + private String commsIdentifier; private boolean handshakeComplete = false; - + private final Logger logger = LoggerFactory.getLogger(SocketClientProtocol.class); - + private Response handshakeResponse = null; private boolean readyForFileTransfer = false; private String transitUriPrefix = null; private int timeoutMillis = 30000; - + private int batchCount; private long batchSize; private long batchMillis; private EventReporter eventReporter; private static final long BATCH_SEND_NANOS = TimeUnit.SECONDS.toNanos(5L); // send batches of up to 5 seconds - + public SocketClientProtocol() { } public void setPreferredBatchCount(final int count) { this.batchCount = count; } - + public void setPreferredBatchSize(final long bytes) { this.batchSize = bytes; } - + public void setPreferredBatchDuration(final long millis) { this.batchMillis = millis; } - + public void setEventReporter(final EventReporter eventReporter) { - this.eventReporter = eventReporter; + this.eventReporter = eventReporter; } - + public void setDestination(final RemoteDestination destination) { this.destination = destination; this.useCompression = destination.isUseCompression(); } - + public void setTimeout(final int timeoutMillis) { - this.timeoutMillis = timeoutMillis; + this.timeoutMillis = timeoutMillis; } - + @Override public void handshake(final Peer peer) throws IOException, HandshakeException { - handshake(peer, destination.getIdentifier()); + handshake(peer, destination.getIdentifier()); } - + public void handshake(final Peer peer, final String destinationId) throws IOException, HandshakeException { - if ( handshakeComplete ) { + if (handshakeComplete) { throw new IllegalStateException("Handshake has already been completed"); } commsIdentifier = UUID.randomUUID().toString(); logger.debug("{} handshaking with {}", this, peer); - + final Map<HandshakeProperty, String> properties = new HashMap<>(); properties.put(HandshakeProperty.GZIP, String.valueOf(useCompression)); - - if ( destinationId != null ) { - properties.put(HandshakeProperty.PORT_IDENTIFIER, destinationId); + + if (destinationId != null) { + properties.put(HandshakeProperty.PORT_IDENTIFIER, destinationId); } - - properties.put(HandshakeProperty.REQUEST_EXPIRATION_MILLIS, String.valueOf(timeoutMillis) ); - - if ( versionNegotiator.getVersion() >= 5 ) { - if ( batchCount > 0 ) { + + properties.put(HandshakeProperty.REQUEST_EXPIRATION_MILLIS, String.valueOf(timeoutMillis)); + + if (versionNegotiator.getVersion() >= 5) { + if (batchCount > 0) { properties.put(HandshakeProperty.BATCH_COUNT, String.valueOf(batchCount)); } - if ( batchSize > 0L ) { + if (batchSize > 0L) { properties.put(HandshakeProperty.BATCH_SIZE, String.valueOf(batchSize)); } - if ( batchMillis > 0L ) { + if (batchMillis > 0L) { properties.put(HandshakeProperty.BATCH_DURATION, String.valueOf(batchMillis)); } } - + final CommunicationsSession commsSession = peer.getCommunicationsSession(); commsSession.setTimeout(timeoutMillis); final DataInputStream dis = new DataInputStream(commsSession.getInput().getInputStream()); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); - + dos.writeUTF(commsIdentifier); - - if ( versionNegotiator.getVersion() >= 3 ) { + + if (versionNegotiator.getVersion() >= 3) { dos.writeUTF(peer.getUrl()); transitUriPrefix = peer.getUrl(); - - if ( !transitUriPrefix.endsWith("/") ) { + + if (!transitUriPrefix.endsWith("/")) { transitUriPrefix = transitUriPrefix + "/"; } } - + logger.debug("Handshaking with properties {}", properties); dos.writeInt(properties.size()); - for ( final Map.Entry<HandshakeProperty, String> entry : properties.entrySet() ) { + for (final Map.Entry<HandshakeProperty, String> entry : properties.entrySet()) { dos.writeUTF(entry.getKey().name()); dos.writeUTF(entry.getValue()); } - + dos.flush(); - + try { handshakeResponse = Response.read(dis); } catch (final ProtocolException e) { throw new HandshakeException(e); } - + switch (handshakeResponse.getCode()) { case PORT_NOT_IN_VALID_STATE: case UNKNOWN_PORT: @@ -181,71 +182,75 @@ public class SocketClientProtocol implements ClientProtocol { readyForFileTransfer = true; break; default: - logger.error("{} received unexpected response {} from {} when negotiating Codec", new Object[] { + logger.error("{} received unexpected response {} from {} when negotiating Codec", new Object[]{ this, handshakeResponse, peer}); peer.close(); throw new HandshakeException("Received unexpected response " + handshakeResponse); } - + logger.debug("{} Finished handshake with {}", this, peer); handshakeComplete = true; } - + + @Override public boolean isReadyForFileTransfer() { return readyForFileTransfer; } - + + @Override public boolean isPortInvalid() { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not completed successfully"); } return handshakeResponse.getCode() == ResponseCode.PORT_NOT_IN_VALID_STATE; } - + + @Override public boolean isPortUnknown() { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not completed successfully"); } return handshakeResponse.getCode() == ResponseCode.UNKNOWN_PORT; } - + + @Override public boolean isDestinationFull() { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not completed successfully"); } return handshakeResponse.getCode() == ResponseCode.PORTS_DESTINATION_FULL; } - + @Override public Set<PeerStatus> getPeerStatuses(final Peer peer) throws IOException { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not been performed"); } - + logger.debug("{} Get Peer Statuses from {}", this, peer); final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataInputStream dis = new DataInputStream(commsSession.getInput().getInputStream()); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); - + RequestType.REQUEST_PEER_LIST.writeRequestType(dos); dos.flush(); final int numPeers = dis.readInt(); final Set<PeerStatus> peers = new HashSet<>(numPeers); - for (int i=0; i < numPeers; i++) { + for (int i = 0; i < numPeers; i++) { final String hostname = dis.readUTF(); final int port = dis.readInt(); final boolean secure = dis.readBoolean(); final int flowFileCount = dis.readInt(); peers.add(new PeerStatus(new PeerDescription(hostname, port, secure), flowFileCount)); } - + logger.debug("{} Received {} Peer Statuses from {}", this, peers.size(), peer); return peers; } - + @Override public FlowFileCodec negotiateCodec(final Peer peer) throws IOException, ProtocolException { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not been performed"); } @@ -255,177 +260,174 @@ public class SocketClientProtocol implements ClientProtocol { final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(dos); - + FlowFileCodec codec = new StandardFlowFileCodec(); try { codec = (FlowFileCodec) RemoteResourceInitiator.initiateResourceNegotiation(codec, dis, dos); } catch (HandshakeException e) { throw new ProtocolException(e.toString()); } - logger.debug("{} negotiated FlowFileCodec {} with {}", new Object[] {this, codec, commsSession}); + logger.debug("{} negotiated FlowFileCodec {} with {}", new Object[]{this, codec, commsSession}); return codec; } - @Override public Transaction startTransaction(final Peer peer, final FlowFileCodec codec, final TransferDirection direction) throws IOException, ProtocolException { - if ( !handshakeComplete ) { + if (!handshakeComplete) { throw new IllegalStateException("Handshake has not been performed"); } - if ( !readyForFileTransfer ) { + if (!readyForFileTransfer) { throw new IllegalStateException("Cannot start transaction; handshake resolution was " + handshakeResponse); } - - return new SocketClientTransaction(versionNegotiator.getVersion(), destination.getIdentifier(), peer, codec, - direction, useCompression, (int) destination.getYieldPeriod(TimeUnit.MILLISECONDS), eventReporter); - } + return new SocketClientTransaction(versionNegotiator.getVersion(), destination.getIdentifier(), peer, codec, + direction, useCompression, (int) destination.getYieldPeriod(TimeUnit.MILLISECONDS), eventReporter); + } @Override public int receiveFlowFiles(final Peer peer, final ProcessContext context, final ProcessSession session, final FlowFileCodec codec) throws IOException, ProtocolException { - final String userDn = peer.getCommunicationsSession().getUserDn(); - final Transaction transaction = startTransaction(peer, codec, TransferDirection.RECEIVE); - - final StopWatch stopWatch = new StopWatch(true); - final Set<FlowFile> flowFilesReceived = new HashSet<>(); - long bytesReceived = 0L; - - while (true) { - final long start = System.nanoTime(); - final DataPacket dataPacket = transaction.receive(); - if ( dataPacket == null ) { - if ( flowFilesReceived.isEmpty() ) { - peer.penalize(destination.getIdentifier(), destination.getYieldPeriod(TimeUnit.MILLISECONDS)); - } - break; - } - - FlowFile flowFile = session.create(); - flowFile = session.putAllAttributes(flowFile, dataPacket.getAttributes()); - flowFile = session.importFrom(dataPacket.getData(), flowFile); - final long receiveNanos = System.nanoTime() - start; - - String sourceFlowFileIdentifier = dataPacket.getAttributes().get(CoreAttributes.UUID.key()); - if ( sourceFlowFileIdentifier == null ) { - sourceFlowFileIdentifier = "<Unknown Identifier>"; - } - - final String transitUri = (transitUriPrefix == null) ? peer.getUrl() : transitUriPrefix + sourceFlowFileIdentifier; - session.getProvenanceReporter().receive(flowFile, transitUri, "urn:nifi:" + sourceFlowFileIdentifier, "Remote Host=" + peer.getHost() + ", Remote DN=" + userDn, TimeUnit.NANOSECONDS.toMillis(receiveNanos)); - - session.transfer(flowFile, Relationship.ANONYMOUS); - bytesReceived += dataPacket.getSize(); - } - - // Confirm that what we received was the correct data. - transaction.confirm(); - - // Commit the session so that we have persisted the data - session.commit(); - - transaction.complete(); - logger.debug("{} Sending TRANSACTION_FINISHED_BUT_DESTINATION_FULL to {}", this, peer); - - if ( !flowFilesReceived.isEmpty() ) { - stopWatch.stop(); - final String flowFileDescription = flowFilesReceived.size() < 20 ? flowFilesReceived.toString() : flowFilesReceived.size() + " FlowFiles"; - final String uploadDataRate = stopWatch.calculateDataRate(bytesReceived); - final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); - final String dataSize = FormatUtils.formatDataSize(bytesReceived); - logger.info("{} Successfully receveied {} ({}) from {} in {} milliseconds at a rate of {}", new Object[] { - this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate }); - } - - return flowFilesReceived.size(); + final String userDn = peer.getCommunicationsSession().getUserDn(); + final Transaction transaction = startTransaction(peer, codec, TransferDirection.RECEIVE); + + final StopWatch stopWatch = new StopWatch(true); + final Set<FlowFile> flowFilesReceived = new HashSet<>(); + long bytesReceived = 0L; + + while (true) { + final long start = System.nanoTime(); + final DataPacket dataPacket = transaction.receive(); + if (dataPacket == null) { + if (flowFilesReceived.isEmpty()) { + peer.penalize(destination.getIdentifier(), destination.getYieldPeriod(TimeUnit.MILLISECONDS)); + } + break; + } + + FlowFile flowFile = session.create(); + flowFile = session.putAllAttributes(flowFile, dataPacket.getAttributes()); + flowFile = session.importFrom(dataPacket.getData(), flowFile); + final long receiveNanos = System.nanoTime() - start; + + String sourceFlowFileIdentifier = dataPacket.getAttributes().get(CoreAttributes.UUID.key()); + if (sourceFlowFileIdentifier == null) { + sourceFlowFileIdentifier = "<Unknown Identifier>"; + } + + final String transitUri = (transitUriPrefix == null) ? peer.getUrl() : transitUriPrefix + sourceFlowFileIdentifier; + session.getProvenanceReporter().receive(flowFile, transitUri, "urn:nifi:" + sourceFlowFileIdentifier, "Remote Host=" + + peer.getHost() + ", Remote DN=" + userDn, TimeUnit.NANOSECONDS.toMillis(receiveNanos)); + + session.transfer(flowFile, Relationship.ANONYMOUS); + bytesReceived += dataPacket.getSize(); + } + + // Confirm that what we received was the correct data. + transaction.confirm(); + + // Commit the session so that we have persisted the data + session.commit(); + + transaction.complete(); + logger.debug("{} Sending TRANSACTION_FINISHED_BUT_DESTINATION_FULL to {}", this, peer); + + if (!flowFilesReceived.isEmpty()) { + stopWatch.stop(); + final String flowFileDescription = flowFilesReceived.size() < 20 ? flowFilesReceived.toString() : flowFilesReceived.size() + " FlowFiles"; + final String uploadDataRate = stopWatch.calculateDataRate(bytesReceived); + final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); + final String dataSize = FormatUtils.formatDataSize(bytesReceived); + logger.info("{} Successfully receveied {} ({}) from {} in {} milliseconds at a rate of {}", new Object[]{ + this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate}); + } + + return flowFilesReceived.size(); } - @Override public int transferFlowFiles(final Peer peer, final ProcessContext context, final ProcessSession session, final FlowFileCodec codec) throws IOException, ProtocolException { - FlowFile flowFile = session.get(); - if (flowFile == null) { - return 0; - } - - try { - final String userDn = peer.getCommunicationsSession().getUserDn(); - final long startSendingNanos = System.nanoTime(); - final StopWatch stopWatch = new StopWatch(true); - long bytesSent = 0L; - - final Transaction transaction = startTransaction(peer, codec, TransferDirection.SEND); - - final Set<FlowFile> flowFilesSent = new HashSet<>(); - boolean continueTransaction = true; - while (continueTransaction) { - final long startNanos = System.nanoTime(); - // call codec.encode within a session callback so that we have the InputStream to read the FlowFile - final FlowFile toWrap = flowFile; - session.read(flowFile, new InputStreamCallback() { - @Override - public void process(final InputStream in) throws IOException { - final DataPacket dataPacket = new StandardDataPacket(toWrap.getAttributes(), in, toWrap.getSize()); - transaction.send(dataPacket); - } - }); - - final long transferNanos = System.nanoTime() - startNanos; - final long transferMillis = TimeUnit.MILLISECONDS.convert(transferNanos, TimeUnit.NANOSECONDS); - - flowFilesSent.add(flowFile); - bytesSent += flowFile.getSize(); - logger.debug("{} Sent {} to {}", this, flowFile, peer); - - final String transitUri = (transitUriPrefix == null) ? peer.getUrl() : transitUriPrefix + flowFile.getAttribute(CoreAttributes.UUID.key()); - session.getProvenanceReporter().send(flowFile, transitUri, "Remote Host=" + peer.getHost() + ", Remote DN=" + userDn, transferMillis, false); - session.remove(flowFile); - - final long sendingNanos = System.nanoTime() - startSendingNanos; - if ( sendingNanos < BATCH_SEND_NANOS ) { - flowFile = session.get(); - } else { - flowFile = null; - } - - continueTransaction = (flowFile != null); - } - - transaction.confirm(); - - // consume input stream entirely, ignoring its contents. If we - // don't do this, the Connection will not be returned to the pool - stopWatch.stop(); - final String uploadDataRate = stopWatch.calculateDataRate(bytesSent); - final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); - final String dataSize = FormatUtils.formatDataSize(bytesSent); - - session.commit(); - transaction.complete(); - - final String flowFileDescription = (flowFilesSent.size() < 20) ? flowFilesSent.toString() : flowFilesSent.size() + " FlowFiles"; - logger.info("{} Successfully sent {} ({}) to {} in {} milliseconds at a rate of {}", new Object[] { - this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate}); - - return flowFilesSent.size(); - } catch (final Exception e) { - session.rollback(); - throw e; - } + FlowFile flowFile = session.get(); + if (flowFile == null) { + return 0; + } + + try { + final String userDn = peer.getCommunicationsSession().getUserDn(); + final long startSendingNanos = System.nanoTime(); + final StopWatch stopWatch = new StopWatch(true); + long bytesSent = 0L; + + final Transaction transaction = startTransaction(peer, codec, TransferDirection.SEND); + + final Set<FlowFile> flowFilesSent = new HashSet<>(); + boolean continueTransaction = true; + while (continueTransaction) { + final long startNanos = System.nanoTime(); + // call codec.encode within a session callback so that we have the InputStream to read the FlowFile + final FlowFile toWrap = flowFile; + session.read(flowFile, new InputStreamCallback() { + @Override + public void process(final InputStream in) throws IOException { + final DataPacket dataPacket = new StandardDataPacket(toWrap.getAttributes(), in, toWrap.getSize()); + transaction.send(dataPacket); + } + }); + + final long transferNanos = System.nanoTime() - startNanos; + final long transferMillis = TimeUnit.MILLISECONDS.convert(transferNanos, TimeUnit.NANOSECONDS); + + flowFilesSent.add(flowFile); + bytesSent += flowFile.getSize(); + logger.debug("{} Sent {} to {}", this, flowFile, peer); + + final String transitUri = (transitUriPrefix == null) ? peer.getUrl() : transitUriPrefix + flowFile.getAttribute(CoreAttributes.UUID.key()); + session.getProvenanceReporter().send(flowFile, transitUri, "Remote Host=" + peer.getHost() + ", Remote DN=" + userDn, transferMillis, false); + session.remove(flowFile); + + final long sendingNanos = System.nanoTime() - startSendingNanos; + if (sendingNanos < BATCH_SEND_NANOS) { + flowFile = session.get(); + } else { + flowFile = null; + } + + continueTransaction = (flowFile != null); + } + + transaction.confirm(); + + // consume input stream entirely, ignoring its contents. If we + // don't do this, the Connection will not be returned to the pool + stopWatch.stop(); + final String uploadDataRate = stopWatch.calculateDataRate(bytesSent); + final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS); + final String dataSize = FormatUtils.formatDataSize(bytesSent); + + session.commit(); + transaction.complete(); + + final String flowFileDescription = (flowFilesSent.size() < 20) ? flowFilesSent.toString() : flowFilesSent.size() + " FlowFiles"; + logger.info("{} Successfully sent {} ({}) to {} in {} milliseconds at a rate of {}", new Object[]{ + this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate}); + + return flowFilesSent.size(); + } catch (final Exception e) { + session.rollback(); + throw e; + } } - - + @Override public VersionNegotiator getVersionNegotiator() { return versionNegotiator; } - + @Override public void shutdown(final Peer peer) throws IOException { readyForFileTransfer = false; final CommunicationsSession commsSession = peer.getCommunicationsSession(); final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream()); - + logger.debug("{} Shutting down with {}", this, peer); // Indicate that we would like to have some data RequestType.SHUTDOWN.writeRequestType(dos); @@ -436,7 +438,7 @@ public class SocketClientProtocol implements ClientProtocol { public String getResourceName() { return "SocketFlowFileProtocol"; } - + @Override public String toString() { return "SocketClientProtocol[CommsID=" + commsIdentifier + "]"; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java index e69104f..e83ea28 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransaction.java @@ -45,50 +45,51 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class SocketClientTransaction implements Transaction { - private static final Logger logger = LoggerFactory.getLogger(SocketClientTransaction.class); - - private final long creationNanoTime = System.nanoTime(); - private final CRC32 crc = new CRC32(); - private final int protocolVersion; - private final FlowFileCodec codec; - private final DataInputStream dis; - private final DataOutputStream dos; - private final TransferDirection direction; - private final boolean compress; - private final Peer peer; - private final int penaltyMillis; - private final String destinationId; - private final EventReporter eventReporter; - - private boolean dataAvailable = false; - private int transfers = 0; - private long contentBytes = 0; - private TransactionState state; - - SocketClientTransaction(final int protocolVersion, final String destinationId, final Peer peer, final FlowFileCodec codec, - final TransferDirection direction, final boolean useCompression, final int penaltyMillis, final EventReporter eventReporter) throws IOException { - this.protocolVersion = protocolVersion; - this.destinationId = destinationId; - this.peer = peer; - this.codec = codec; - this.direction = direction; - this.dis = new DataInputStream(peer.getCommunicationsSession().getInput().getInputStream()); - this.dos = new DataOutputStream(peer.getCommunicationsSession().getOutput().getOutputStream()); - this.compress = useCompression; - this.state = TransactionState.TRANSACTION_STARTED; - this.penaltyMillis = penaltyMillis; - this.eventReporter = eventReporter; - - initialize(); - } - - private void initialize() throws IOException { - try { - if ( direction == TransferDirection.RECEIVE ) { + + private static final Logger logger = LoggerFactory.getLogger(SocketClientTransaction.class); + + private final long creationNanoTime = System.nanoTime(); + private final CRC32 crc = new CRC32(); + private final int protocolVersion; + private final FlowFileCodec codec; + private final DataInputStream dis; + private final DataOutputStream dos; + private final TransferDirection direction; + private final boolean compress; + private final Peer peer; + private final int penaltyMillis; + private final String destinationId; + private final EventReporter eventReporter; + + private boolean dataAvailable = false; + private int transfers = 0; + private long contentBytes = 0; + private TransactionState state; + + SocketClientTransaction(final int protocolVersion, final String destinationId, final Peer peer, final FlowFileCodec codec, + final TransferDirection direction, final boolean useCompression, final int penaltyMillis, final EventReporter eventReporter) throws IOException { + this.protocolVersion = protocolVersion; + this.destinationId = destinationId; + this.peer = peer; + this.codec = codec; + this.direction = direction; + this.dis = new DataInputStream(peer.getCommunicationsSession().getInput().getInputStream()); + this.dos = new DataOutputStream(peer.getCommunicationsSession().getOutput().getOutputStream()); + this.compress = useCompression; + this.state = TransactionState.TRANSACTION_STARTED; + this.penaltyMillis = penaltyMillis; + this.eventReporter = eventReporter; + + initialize(); + } + + private void initialize() throws IOException { + try { + if (direction == TransferDirection.RECEIVE) { // Indicate that we would like to have some data RequestType.RECEIVE_FLOWFILES.writeRequestType(dos); dos.flush(); - + final Response dataAvailableCode = Response.read(dis); switch (dataAvailableCode.getCode()) { case MORE_DATA: @@ -102,39 +103,38 @@ public class SocketClientTransaction implements Transaction { default: throw new ProtocolException("Got unexpected response when asking for data: " + dataAvailableCode); } - + } else { // Indicate that we would like to have some data RequestType.SEND_FLOWFILES.writeRequestType(dos); dos.flush(); } - } catch (final Exception e) { - error(); - throw e; - } - } - - - @Override - public DataPacket receive() throws IOException { - try { - try { - if ( state != TransactionState.DATA_EXCHANGED && state != TransactionState.TRANSACTION_STARTED) { - throw new IllegalStateException("Cannot receive data from " + peer + " because Transaction State is " + state); - } - - if ( direction == TransferDirection.SEND ) { - throw new IllegalStateException("Attempting to receive data from " + peer + " but started a SEND Transaction"); - } - - // if we already know there's no data, just return null - if ( !dataAvailable ) { - return null; - } - - // if we have already received a packet, check if another is available. - if ( transfers > 0 ) { - // Determine if Peer will send us data or has no data to send us + } catch (final Exception e) { + error(); + throw e; + } + } + + @Override + public DataPacket receive() throws IOException { + try { + try { + if (state != TransactionState.DATA_EXCHANGED && state != TransactionState.TRANSACTION_STARTED) { + throw new IllegalStateException("Cannot receive data from " + peer + " because Transaction State is " + state); + } + + if (direction == TransferDirection.SEND) { + throw new IllegalStateException("Attempting to receive data from " + peer + " but started a SEND Transaction"); + } + + // if we already know there's no data, just return null + if (!dataAvailable) { + return null; + } + + // if we have already received a packet, check if another is available. + if (transfers > 0) { + // Determine if Peer will send us data or has no data to send us final Response dataAvailableCode = Response.read(dis); switch (dataAvailableCode.getCode()) { case CONTINUE_TRANSACTION: @@ -149,170 +149,166 @@ public class SocketClientTransaction implements Transaction { throw new ProtocolException("Got unexpected response from " + peer + " when asking for data: " + dataAvailableCode); } } - - // if no data available, return null - if ( !dataAvailable ) { - return null; - } - + + // if no data available, return null + if (!dataAvailable) { + return null; + } + logger.debug("{} Receiving data from {}", this, peer); final InputStream dataIn = compress ? new CompressionInputStream(dis) : dis; final DataPacket packet = codec.decode(new CheckedInputStream(dataIn, crc)); - - if ( packet == null ) { + + if (packet == null) { this.dataAvailable = false; } else { - transfers++; - contentBytes += packet.getSize(); + transfers++; + contentBytes += packet.getSize(); } - + this.state = TransactionState.DATA_EXCHANGED; return packet; - } catch (final IOException ioe) { - throw new IOException("Failed to receive data from " + peer + " due to " + ioe, ioe); - } - } catch (final Exception e) { - error(); - throw e; - } - } - - - @Override - public void send(final byte[] content, final Map<String, String> attributes) throws IOException { - send(new StandardDataPacket(attributes, new ByteArrayInputStream(content), content.length)); - } - - @Override - public void send(final DataPacket dataPacket) throws IOException { - try { - try { - if ( state != TransactionState.DATA_EXCHANGED && state != TransactionState.TRANSACTION_STARTED) { - throw new IllegalStateException("Cannot send data to " + peer + " because Transaction State is " + state); - } - - if ( direction == TransferDirection.RECEIVE ) { + } catch (final IOException ioe) { + throw new IOException("Failed to receive data from " + peer + " due to " + ioe, ioe); + } + } catch (final Exception e) { + error(); + throw e; + } + } + + @Override + public void send(final byte[] content, final Map<String, String> attributes) throws IOException { + send(new StandardDataPacket(attributes, new ByteArrayInputStream(content), content.length)); + } + + @Override + public void send(final DataPacket dataPacket) throws IOException { + try { + try { + if (state != TransactionState.DATA_EXCHANGED && state != TransactionState.TRANSACTION_STARTED) { + throw new IllegalStateException("Cannot send data to " + peer + " because Transaction State is " + state); + } + + if (direction == TransferDirection.RECEIVE) { throw new IllegalStateException("Attempting to send data to " + peer + " but started a RECEIVE Transaction"); } - - if ( transfers > 0 ) { + + if (transfers > 0) { ResponseCode.CONTINUE_TRANSACTION.writeResponse(dos); } - + logger.debug("{} Sending data to {}", this, peer); - + final OutputStream dataOut = compress ? new CompressionOutputStream(dos) : dos; - final OutputStream out = new CheckedOutputStream(dataOut, crc); + final OutputStream out = new CheckedOutputStream(dataOut, crc); codec.encode(dataPacket, out); - + // need to close the CompressionOutputStream in order to force it write out any remaining bytes. // Otherwise, do NOT close it because we don't want to close the underlying stream // (CompressionOutputStream will not close the underlying stream when it's closed) - if ( compress ) { - out.close(); + if (compress) { + out.close(); } - + transfers++; contentBytes += dataPacket.getSize(); this.state = TransactionState.DATA_EXCHANGED; - } catch (final IOException ioe) { - throw new IOException("Failed to send data to " + peer + " due to " + ioe, ioe); - } - } catch (final Exception e) { - error(); - throw e; - } - } - - - @Override - public void cancel(final String explanation) throws IOException { - if ( state == TransactionState.TRANSACTION_CANCELED || state == TransactionState.TRANSACTION_COMPLETED || state == TransactionState.ERROR ) { - throw new IllegalStateException("Cannot cancel transaction because state is already " + state); - } - - try { - ResponseCode.CANCEL_TRANSACTION.writeResponse(dos, explanation == null ? "<No explanation given>" : explanation); - state = TransactionState.TRANSACTION_CANCELED; - } catch (final IOException ioe) { - error(); - throw new IOException("Failed to send 'cancel transaction' message to " + peer + " due to " + ioe, ioe); - } - } - - - @Override - public TransactionCompletion complete() throws IOException { - try { - try { - if ( state != TransactionState.TRANSACTION_CONFIRMED ) { - throw new IllegalStateException("Cannot complete transaction with " + peer + " because state is " + state + - "; Transaction can only be completed when state is " + TransactionState.TRANSACTION_CONFIRMED); - } - - boolean backoff = false; - if ( direction == TransferDirection.RECEIVE ) { - if ( transfers == 0 ) { - state = TransactionState.TRANSACTION_COMPLETED; - return new SocketClientTransactionCompletion(false, 0, 0L, System.nanoTime() - creationNanoTime); - } - + } catch (final IOException ioe) { + throw new IOException("Failed to send data to " + peer + " due to " + ioe, ioe); + } + } catch (final Exception e) { + error(); + throw e; + } + } + + @Override + public void cancel(final String explanation) throws IOException { + if (state == TransactionState.TRANSACTION_CANCELED || state == TransactionState.TRANSACTION_COMPLETED || state == TransactionState.ERROR) { + throw new IllegalStateException("Cannot cancel transaction because state is already " + state); + } + + try { + ResponseCode.CANCEL_TRANSACTION.writeResponse(dos, explanation == null ? "<No explanation given>" : explanation); + state = TransactionState.TRANSACTION_CANCELED; + } catch (final IOException ioe) { + error(); + throw new IOException("Failed to send 'cancel transaction' message to " + peer + " due to " + ioe, ioe); + } + } + + @Override + public TransactionCompletion complete() throws IOException { + try { + try { + if (state != TransactionState.TRANSACTION_CONFIRMED) { + throw new IllegalStateException("Cannot complete transaction with " + peer + " because state is " + state + + "; Transaction can only be completed when state is " + TransactionState.TRANSACTION_CONFIRMED); + } + + boolean backoff = false; + if (direction == TransferDirection.RECEIVE) { + if (transfers == 0) { + state = TransactionState.TRANSACTION_COMPLETED; + return new SocketClientTransactionCompletion(false, 0, 0L, System.nanoTime() - creationNanoTime); + } + // Confirm that we received the data and the peer can now discard it logger.debug("{} Sending TRANSACTION_FINISHED to {}", this, peer); ResponseCode.TRANSACTION_FINISHED.writeResponse(dos); - + state = TransactionState.TRANSACTION_COMPLETED; } else { final Response transactionResponse; try { transactionResponse = Response.read(dis); } catch (final IOException e) { - throw new IOException(this + " Failed to receive a response from " + peer + " when expecting a TransactionFinished Indicator. " + - "It is unknown whether or not the peer successfully received/processed the data.", e); + throw new IOException(this + " Failed to receive a response from " + peer + " when expecting a TransactionFinished Indicator. " + + "It is unknown whether or not the peer successfully received/processed the data.", e); } - + logger.debug("{} Received {} from {}", this, transactionResponse, peer); - if ( transactionResponse.getCode() == ResponseCode.TRANSACTION_FINISHED_BUT_DESTINATION_FULL ) { + if (transactionResponse.getCode() == ResponseCode.TRANSACTION_FINISHED_BUT_DESTINATION_FULL) { peer.penalize(destinationId, penaltyMillis); backoff = true; - } else if ( transactionResponse.getCode() != ResponseCode.TRANSACTION_FINISHED ) { + } else if (transactionResponse.getCode() != ResponseCode.TRANSACTION_FINISHED) { throw new ProtocolException("After sending data to " + peer + ", expected TRANSACTION_FINISHED response but got " + transactionResponse); } - + state = TransactionState.TRANSACTION_COMPLETED; } - - return new SocketClientTransactionCompletion(backoff, transfers, contentBytes, System.nanoTime() - creationNanoTime); - } catch (final IOException ioe) { - throw new IOException("Failed to complete transaction with " + peer + " due to " + ioe, ioe); - } - } catch (final Exception e) { - error(); - throw e; - } - } - - - @Override - public void confirm() throws IOException { - try { - try { - if ( state == TransactionState.TRANSACTION_STARTED && !dataAvailable && direction == TransferDirection.RECEIVE ) { - // client requested to receive data but no data available. no need to confirm. - state = TransactionState.TRANSACTION_CONFIRMED; - return; - } - - if ( state != TransactionState.DATA_EXCHANGED ) { - throw new IllegalStateException("Cannot confirm Transaction because state is " + state + - "; Transaction can only be confirmed when state is " + TransactionState.DATA_EXCHANGED ); - } - - if ( direction == TransferDirection.RECEIVE ) { - if ( dataAvailable ) { + + return new SocketClientTransactionCompletion(backoff, transfers, contentBytes, System.nanoTime() - creationNanoTime); + } catch (final IOException ioe) { + throw new IOException("Failed to complete transaction with " + peer + " due to " + ioe, ioe); + } + } catch (final Exception e) { + error(); + throw e; + } + } + + @Override + public void confirm() throws IOException { + try { + try { + if (state == TransactionState.TRANSACTION_STARTED && !dataAvailable && direction == TransferDirection.RECEIVE) { + // client requested to receive data but no data available. no need to confirm. + state = TransactionState.TRANSACTION_CONFIRMED; + return; + } + + if (state != TransactionState.DATA_EXCHANGED) { + throw new IllegalStateException("Cannot confirm Transaction because state is " + state + + "; Transaction can only be confirmed when state is " + TransactionState.DATA_EXCHANGED); + } + + if (direction == TransferDirection.RECEIVE) { + if (dataAvailable) { throw new IllegalStateException("Cannot complete transaction because the sender has already sent more data than client has consumed."); } - + // we received a FINISH_TRANSACTION indicator. Send back a CONFIRM_TRANSACTION message // to peer so that we can verify that the connection is still open. This is a two-phase commit, // which helps to prevent the chances of data duplication. Without doing this, we may commit the @@ -323,84 +319,88 @@ public class SocketClientTransaction implements Transaction { logger.trace("{} Sending CONFIRM_TRANSACTION Response Code to {}", this, peer); final String calculatedCRC = String.valueOf(crc.getValue()); ResponseCode.CONFIRM_TRANSACTION.writeResponse(dos, calculatedCRC); - + final Response confirmTransactionResponse; try { confirmTransactionResponse = Response.read(dis); } catch (final IOException ioe) { logger.error("Failed to receive response code from {} when expecting confirmation of transaction", peer); - if ( eventReporter != null ) { - eventReporter.reportEvent(Severity.ERROR, "Site-to-Site", "Failed to receive response code from " + peer + " when expecting confirmation of transaction"); + if (eventReporter != null) { + eventReporter.reportEvent(Severity.ERROR, "Site-to-Site", "Failed to receive response code from " + peer + " when expecting confirmation of transaction"); } throw ioe; } - + logger.trace("{} Received {} from {}", this, confirmTransactionResponse, peer); - + switch (confirmTransactionResponse.getCode()) { case CONFIRM_TRANSACTION: break; case BAD_CHECKSUM: throw new IOException(this + " Received a BadChecksum response from peer " + peer); default: - throw new ProtocolException(this + " Received unexpected Response from peer " + peer + " : " + confirmTransactionResponse + "; expected 'Confirm Transaction' Response Code"); + throw new ProtocolException(this + " Received unexpected Response from peer " + peer + " : " + + confirmTransactionResponse + "; expected 'Confirm Transaction' Response Code"); } - + state = TransactionState.TRANSACTION_CONFIRMED; } else { logger.debug("{} Sent FINISH_TRANSACTION indicator to {}", this, peer); ResponseCode.FINISH_TRANSACTION.writeResponse(dos); - + final String calculatedCRC = String.valueOf(crc.getValue()); - + // we've sent a FINISH_TRANSACTION. Now we'll wait for the peer to send a 'Confirm Transaction' response final Response transactionConfirmationResponse = Response.read(dis); - if ( transactionConfirmationResponse.getCode() == ResponseCode.CONFIRM_TRANSACTION ) { + if (transactionConfirmationResponse.getCode() == ResponseCode.CONFIRM_TRANSACTION) { // Confirm checksum and echo back the confirmation. logger.trace("{} Received {} from {}", this, transactionConfirmationResponse, peer); final String receivedCRC = transactionConfirmationResponse.getMessage(); - + // CRC was not used before version 4 - if ( protocolVersion > 3 ) { - if ( !receivedCRC.equals(calculatedCRC) ) { + if (protocolVersion > 3) { + if (!receivedCRC.equals(calculatedCRC)) { ResponseCode.BAD_CHECKSUM.writeResponse(dos); - throw new IOException(this + " Sent data to peer " + peer + " but calculated CRC32 Checksum as " + calculatedCRC + " while peer calculated CRC32 Checksum as " + receivedCRC + "; canceling transaction and rolling back session"); + throw new IOException(this + " Sent data to peer " + peer + " but calculated CRC32 Checksum as " + + calculatedCRC + " while peer calculated CRC32 Checksum as " + + receivedCRC + "; canceling transaction and rolling back session"); } } - + ResponseCode.CONFIRM_TRANSACTION.writeResponse(dos, ""); } else { - throw new ProtocolException("Expected to receive 'Confirm Transaction' response from peer " + peer + " but received " + transactionConfirmationResponse); + throw new ProtocolException("Expected to receive 'Confirm Transaction' response from peer " + + peer + " but received " + transactionConfirmationResponse); } - + state = TransactionState.TRANSACTION_CONFIRMED; } - } catch (final IOException ioe) { - throw new IOException("Failed to confirm transaction with " + peer + " due to " + ioe, ioe); - } - } catch (final Exception e) { - error(); - throw e; - } - } - - @Override - public void error() { - this.state = TransactionState.ERROR; - } - - @Override - public TransactionState getState() { - return state; - } - - @Override - public Communicant getCommunicant() { - return peer; - } - + } catch (final IOException ioe) { + throw new IOException("Failed to confirm transaction with " + peer + " due to " + ioe, ioe); + } + } catch (final Exception e) { + error(); + throw e; + } + } + + @Override + public void error() { + this.state = TransactionState.ERROR; + } + + @Override + public TransactionState getState() { + return state; + } + + @Override + public Communicant getCommunicant() { + return peer; + } + @Override public String toString() { - return "SocketClientTransaction[Url=" + peer.getUrl() + ", TransferDirection=" + direction + ", State=" + state + "]"; + return "SocketClientTransaction[Url=" + peer.getUrl() + ", TransferDirection=" + direction + ", State=" + state + "]"; } } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java index 5eb6c91..bd95013 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/protocol/socket/SocketClientTransactionCompletion.java @@ -26,7 +26,7 @@ public class SocketClientTransactionCompletion implements TransactionCompletion private final int dataPacketsTransferred; private final long bytesTransferred; private final long durationNanos; - + public SocketClientTransactionCompletion(final boolean backoff, final int dataPacketsTransferred, final long bytesTransferred, final long durationNanos) { this.backoff = backoff; this.dataPacketsTransferred = dataPacketsTransferred; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java index 10352ec..d746abf 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/NiFiRestApiUtil.java @@ -32,43 +32,44 @@ import org.codehaus.jackson.JsonNode; import org.codehaus.jackson.map.ObjectMapper; public class NiFiRestApiUtil { + public static final int RESPONSE_CODE_OK = 200; - + private final SSLContext sslContext; - + public NiFiRestApiUtil(final SSLContext sslContext) { this.sslContext = sslContext; } - + private HttpURLConnection getConnection(final String connUrl, final int timeoutMillis) throws IOException { final URL url = new URL(connUrl); final HttpURLConnection connection = (HttpURLConnection) url.openConnection(); connection.setConnectTimeout(timeoutMillis); connection.setReadTimeout(timeoutMillis); - + // special handling for https if (sslContext != null && connection instanceof HttpsURLConnection) { HttpsURLConnection secureConnection = (HttpsURLConnection) connection; secureConnection.setSSLSocketFactory(sslContext.getSocketFactory()); // check the trusted hostname property and override the HostnameVerifier - secureConnection.setHostnameVerifier(new OverrideHostnameVerifier(url.getHost(), + secureConnection.setHostnameVerifier(new OverrideHostnameVerifier(url.getHost(), secureConnection.getHostnameVerifier())); } - + return connection; } - + public ControllerDTO getController(final String url, final int timeoutMillis) throws IOException { final HttpURLConnection connection = getConnection(url, timeoutMillis); connection.setRequestMethod("GET"); final int responseCode = connection.getResponseCode(); - + final ByteArrayOutputStream baos = new ByteArrayOutputStream(); StreamUtils.copy(connection.getInputStream(), baos); final String responseMessage = baos.toString(); - - if ( responseCode == RESPONSE_CODE_OK ) { + + if (responseCode == RESPONSE_CODE_OK) { final ObjectMapper mapper = new ObjectMapper(); final JsonNode jsonNode = mapper.readTree(responseMessage); final JsonNode controllerNode = jsonNode.get("controller"); @@ -77,8 +78,9 @@ public class NiFiRestApiUtil { throw new IOException("Got HTTP response Code " + responseCode + ": " + connection.getResponseMessage() + " with explanation: " + responseMessage); } } - + private static class OverrideHostnameVerifier implements HostnameVerifier { + private final String trustedHostname; private final HostnameVerifier delegate; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java index 6dab77b..c52b4b7 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/PeerStatusCache.java @@ -21,7 +21,8 @@ import java.util.Set; import org.apache.nifi.remote.PeerStatus; public class PeerStatusCache { - private final Set<PeerStatus> statuses; + + private final Set<PeerStatus> statuses; private final long timestamp; public PeerStatusCache(final Set<PeerStatus> statuses) { http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java index bd1b50c..70bb324 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/main/java/org/apache/nifi/remote/util/StandardDataPacket.java @@ -25,26 +25,26 @@ import org.apache.nifi.stream.io.MinimumLengthInputStream; public class StandardDataPacket implements DataPacket { - private final Map<String, String> attributes; - private final InputStream stream; - private final long size; - - public StandardDataPacket(final Map<String, String> attributes, final InputStream stream, final long size) { - this.attributes = attributes; - this.stream = new MinimumLengthInputStream(new LimitingInputStream(stream, size), size); - this.size = size; - } - - public Map<String, String> getAttributes() { - return attributes; - } - - public InputStream getData() { - return stream; - } - - public long getSize() { - return size; - } - + private final Map<String, String> attributes; + private final InputStream stream; + private final long size; + + public StandardDataPacket(final Map<String, String> attributes, final InputStream stream, final long size) { + this.attributes = attributes; + this.stream = new MinimumLengthInputStream(new LimitingInputStream(stream, size), size); + this.size = size; + } + + public Map<String, String> getAttributes() { + return attributes; + } + + public InputStream getData() { + return stream; + } + + public long getSize() { + return size; + } + } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java b/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java index c5cca78..8336745 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestEndpointConnectionStatePool.java @@ -40,11 +40,11 @@ public class TestEndpointConnectionStatePool { clusterNodeInfo.setNodeInformation(collection); final List<PeerStatus> destinations = EndpointConnectionPool.formulateDestinationList(clusterNodeInfo, TransferDirection.RECEIVE); - for ( final PeerStatus peerStatus : destinations ) { + for (final PeerStatus peerStatus : destinations) { System.out.println(peerStatus.getPeerDescription()); } } - + @Test public void testFormulateDestinationListForOutputHugeDifference() throws IOException { final ClusterNodeInformation clusterNodeInfo = new ClusterNodeInformation(); @@ -54,14 +54,11 @@ public class TestEndpointConnectionStatePool { clusterNodeInfo.setNodeInformation(collection); final List<PeerStatus> destinations = EndpointConnectionPool.formulateDestinationList(clusterNodeInfo, TransferDirection.RECEIVE); - for ( final PeerStatus peerStatus : destinations ) { + for (final PeerStatus peerStatus : destinations) { System.out.println(peerStatus.getPeerDescription()); } } - - - - + @Test public void testFormulateDestinationListForInputPorts() throws IOException { final ClusterNodeInformation clusterNodeInfo = new ClusterNodeInformation(); @@ -74,11 +71,11 @@ public class TestEndpointConnectionStatePool { clusterNodeInfo.setNodeInformation(collection); final List<PeerStatus> destinations = EndpointConnectionPool.formulateDestinationList(clusterNodeInfo, TransferDirection.SEND); - for ( final PeerStatus peerStatus : destinations ) { + for (final PeerStatus peerStatus : destinations) { System.out.println(peerStatus.getPeerDescription()); } } - + @Test public void testFormulateDestinationListForInputPortsHugeDifference() throws IOException { final ClusterNodeInformation clusterNodeInfo = new ClusterNodeInformation(); @@ -88,7 +85,7 @@ public class TestEndpointConnectionStatePool { clusterNodeInfo.setNodeInformation(collection); final List<PeerStatus> destinations = EndpointConnectionPool.formulateDestinationList(clusterNodeInfo, TransferDirection.SEND); - for ( final PeerStatus peerStatus : destinations ) { + for (final PeerStatus peerStatus : destinations) { System.out.println(peerStatus.getPeerDescription()); } } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java b/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java index b73e44d..155fc95 100644 --- a/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java +++ b/nifi/nifi-commons/nifi-site-to-site-client/src/test/java/org/apache/nifi/remote/client/socket/TestSiteToSiteClient.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.io.InputStream; import java.util.HashMap; import java.util.Map; -import java.util.concurrent.TimeUnit; import org.apache.nifi.remote.Transaction; import org.apache.nifi.remote.TransferDirection; @@ -39,32 +38,32 @@ public class TestSiteToSiteClient { @Ignore("For local testing only; not really a unit test but a manual test") public void testReceive() throws IOException { System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.remote", "DEBUG"); - + final SiteToSiteClient client = new SiteToSiteClient.Builder() - .url("http://localhost:8080/nifi";) - .portName("cba") - .requestBatchCount(10) - .build(); - + .url("http://localhost:8080/nifi";) + .portName("cba") + .requestBatchCount(10) + .build(); + try { - for (int i=0; i < 1000; i++) { + for (int i = 0; i < 1000; i++) { final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE); Assert.assertNotNull(transaction); - + DataPacket packet; while (true) { packet = transaction.receive(); - if ( packet == null ) { + if (packet == null) { break; } final InputStream in = packet.getData(); final long size = packet.getSize(); final byte[] buff = new byte[(int) size]; - + StreamUtils.fillBuffer(in, buff); } - + transaction.confirm(); transaction.complete(); } @@ -72,34 +71,33 @@ public class TestSiteToSiteClient { client.close(); } } - - + @Test @Ignore("For local testing only; not really a unit test but a manual test") public void testSend() throws IOException { System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.remote", "DEBUG"); - + final SiteToSiteClient client = new SiteToSiteClient.Builder() - .url("http://localhost:8080/nifi";) - .portName("input") - .build(); - + .url("http://localhost:8080/nifi";) + .portName("input") + .build(); + try { final Transaction transaction = client.createTransaction(TransferDirection.SEND); Assert.assertNotNull(transaction); - + final Map<String, String> attrs = new HashMap<>(); attrs.put("site-to-site", "yes, please!"); final byte[] bytes = "Hello".getBytes(); final ByteArrayInputStream bais = new ByteArrayInputStream(bytes); final DataPacket packet = new StandardDataPacket(attrs, bais, bytes.length); transaction.send(packet); - + transaction.confirm(); transaction.complete(); } finally { client.close(); } } - + } http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/AbstractChannelReader.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/AbstractChannelReader.java b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/AbstractChannelReader.java index 172c593..cc24575 100644 --- a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/AbstractChannelReader.java +++ b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/AbstractChannelReader.java @@ -33,10 +33,6 @@ import org.apache.commons.lang3.builder.ToStringStyle; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** - * - * @author none - */ public abstract class AbstractChannelReader implements Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(AbstractChannelReader.class); @@ -91,12 +87,12 @@ public abstract class AbstractChannelReader implements Runnable { * Allows a subclass to specifically handle how it reads from the given * key's channel into the given buffer. * - * @param key - * @param buffer + * @param key of channel to read from + * @param buffer to fill * @return the number of bytes read in the final read cycle. A value of zero * or more indicates the channel is still open but a value of -1 indicates * end of stream. - * @throws IOException + * @throws IOException if reading from channel causes failure */ protected abstract int fillBuffer(SelectionKey key, ByteBuffer buffer) throws IOException; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/BufferPool.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/BufferPool.java b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/BufferPool.java index a413ad2..007034b 100644 --- a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/BufferPool.java +++ b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/BufferPool.java @@ -25,10 +25,6 @@ import java.util.concurrent.LinkedBlockingDeque; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** - * - * @author none - */ public class BufferPool implements Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(BufferPool.class); @@ -50,9 +46,9 @@ public class BufferPool implements Runnable { /** * Returns the given buffer to the pool - and clears it. * - * @param buffer - * @param bytesProcessed - * @return + * @param buffer buffer to return + * @param bytesProcessed bytes processed for this buffer being returned + * @return true if buffer returned to pool */ public synchronized boolean returnBuffer(ByteBuffer buffer, final int bytesProcessed) { totalBytesExtracted += bytesProcessed; http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelDispatcher.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelDispatcher.java b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelDispatcher.java index 2ae2c07..824f2df 100644 --- a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelDispatcher.java +++ b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelDispatcher.java @@ -35,10 +35,6 @@ import org.apache.nifi.io.nio.consumer.StreamConsumerFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** - * - * @author none - */ public final class ChannelDispatcher implements Runnable { private static final Logger LOGGER = LoggerFactory.getLogger(ChannelDispatcher.class); @@ -81,8 +77,8 @@ public final class ChannelDispatcher implements Runnable { /* * When serverSocketsChannels are registered with the selector, want each invoke of this method to loop through all * channels' keys. - * - * @throws IOException + * + * @throws IOException if unable to select keys */ private void selectServerSocketKeys() throws IOException { int numSelected = serverSocketSelector.select(timeout); @@ -121,8 +117,8 @@ public final class ChannelDispatcher implements Runnable { * When invoking this method, only want to iterate through the selected keys once. When a key is entered into the selectors * selected key set, select will return a positive value. The next select will return 0 if nothing has changed. Note that * the selected key set is not manually changed via a remove operation. - * - * @throws IOException + * + * @throws IOException if unable to select keys */ private void selectSocketChannelKeys() throws IOException { // once a channel associated with a key in this selector is 'ready', it causes this select to immediately return. @@ -138,7 +134,7 @@ public final class ChannelDispatcher implements Runnable { // there are 2 kinds of channels in this selector, both which have their own readers and are executed in their own // threads. We will get here whenever a new SocketChannel is created due to an incoming connection. However, // for a DatagramChannel we don't want to create a new reader unless it is a new DatagramChannel. The only - // way to tell if it's new is the lack of an attachment. + // way to tell if it's new is the lack of an attachment. if (channel instanceof DatagramChannel && socketChannelKey.attachment() == null) { reader = new DatagramChannelReader(UUID.randomUUID().toString(), socketChannelKey, emptyBuffers, factory); socketChannelKey.attach(reader); http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/b612b6bc/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelListener.java ---------------------------------------------------------------------- diff --git a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelListener.java b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelListener.java index b0a1cfb..7cbf589 100644 --- a/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelListener.java +++ b/nifi/nifi-commons/nifi-socket-utils/src/main/java/org/apache/nifi/io/nio/ChannelListener.java @@ -62,7 +62,6 @@ import org.slf4j.LoggerFactory; * All ChannelReaders will get throttled by the unavailability of buffers in the * provided BufferPool. This is designed to create back pressure. * - * @author none */ public final class ChannelListener { @@ -99,7 +98,7 @@ public final class ChannelListener { * @param port - port to bind to * @param receiveBufferSize - size of OS receive buffer to request. If less * than 0 then will not be set and OS default will win. - * @throws IOException + * @throws IOException if unable to add socket */ public void addServerSocket(final InetAddress nicIPAddress, final int port, final int receiveBufferSize) throws IOException { @@ -129,7 +128,7 @@ public final class ChannelListener { * @param port - the port to listen on * @param receiveBufferSize - the number of bytes to request for a receive * buffer from OS - * @throws IOException + * @throws IOException if unable to add channel */ public void addDatagramChannel(final InetAddress nicIPAddress, final int port, final int receiveBufferSize) throws IOException { @@ -156,7 +155,7 @@ public final class ChannelListener { * any network interface on the local host. * @param sendingPort - the port used by the sender of datagrams. Only * datagrams from this port will be received. - * @throws IOException + * @throws IOException if unable to add channel */ public void addDatagramChannel(final InetAddress nicIPAddress, final int port, final int receiveBufferSize, final String sendingHost, final Integer sendingPort) throws IOException {
