You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/12/30 10:36:49 UTC
[1/6] camel git commit: Added camel-mllp component
Repository: camel
Updated Branches:
refs/heads/master c34db42c8 -> 0fb50fcf5
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpServerResource.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpServerResource.java b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpServerResource.java
new file mode 100644
index 0000000..c148310
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpServerResource.java
@@ -0,0 +1,878 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.junit.rule.mllp;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.regex.Pattern;
+
+import org.junit.rules.ExternalResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_BLOCK;
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_DATA;
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_STREAM;
+import static org.apache.camel.component.mllp.MllpEndpoint.MESSAGE_TERMINATOR;
+import static org.apache.camel.component.mllp.MllpEndpoint.SEGMENT_DELIMITER;
+import static org.apache.camel.component.mllp.MllpEndpoint.START_OF_BLOCK;
+/**
+ * MLLP Test Server packaged as a JUnit Rule
+ *
+ * The server can be configured to simulate a large number
+ * of error conditions.
+ *
+ * TODO: This needs to be looked at - it may be orphaning threads
+ */
+public class MllpServerResource extends ExternalResource {
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ int listenPort;
+ int backlog = 5;
+
+ int counter = 1;
+
+ boolean active = true;
+
+ int excludeStartOfBlockModulus;
+ int excludeEndOfBlockModulus;
+ int excludeEndOfDataModulus;
+
+ int excludeAcknowledgementModulus;
+
+ int sendOutOfBandDataModulus;
+
+ int disconnectBeforeAcknowledgementModulus;
+ int disconnectAfterAcknowledgementModulus;
+
+ int sendApplicationRejectAcknowledgementModulus;
+ int sendApplicationErrorAcknowledgementModulus;
+
+ Pattern sendApplicationRejectAcknowledgementPattern;
+ Pattern sendApplicationErrorAcknowledgementPattern;
+
+ ServerSocketThread serverSocketThread;
+
+ public MllpServerResource() {
+ }
+
+ public MllpServerResource(int listenPort) {
+ this.listenPort = listenPort;
+ }
+
+ public MllpServerResource(int listenPort, int backlog) {
+ this.listenPort = listenPort;
+ this.backlog = backlog;
+ }
+
+ public int getListenPort() {
+ return listenPort;
+ }
+
+ public void setListenPort(int listenPort) {
+ this.listenPort = listenPort;
+ }
+
+ public int getBacklog() {
+ return backlog;
+ }
+
+ public void setBacklog(int backlog) {
+ this.backlog = backlog;
+ }
+
+ public void startup() throws IOException {
+ this.active = true;
+ serverSocketThread = new ServerSocketThread(listenPort, backlog);
+ if (0 >= listenPort) {
+ listenPort = serverSocketThread.listenPort;
+ }
+ serverSocketThread.setDaemon(true);
+ serverSocketThread.start();
+ }
+
+ public void shutdown() {
+ this.active = false;
+ serverSocketThread.shutdown();
+ serverSocketThread = null;
+ }
+
+
+ @Override
+ protected void before() throws Throwable {
+ startup();
+ super.before();
+ }
+
+ @Override
+ protected void after() {
+ super.after();
+ shutdown();
+ }
+
+ public void interrupt() {
+ serverSocketThread.interrupt();
+ }
+
+ public boolean sendApplicationRejectAcknowledgement(String hl7Message) {
+ return evaluatePatten(hl7Message, this.sendApplicationErrorAcknowledgementPattern);
+ }
+
+ public boolean sendApplicationErrorAcknowledgement(String hl7Message) {
+ return evaluatePatten(hl7Message, this.sendApplicationRejectAcknowledgementPattern);
+ }
+
+ public boolean sendApplicationRejectAcknowledgement(int messageCount) {
+ return evaluateModulus(messageCount, this.sendApplicationRejectAcknowledgementModulus);
+ }
+
+ public boolean sendApplicationErrorAcknowledgement(int messageCount) {
+ return evaluateModulus(messageCount, this.sendApplicationErrorAcknowledgementModulus);
+ }
+
+
+ public boolean excludeStartOfBlock(int messageCount) {
+ return evaluateModulus(messageCount, excludeStartOfBlockModulus);
+ }
+
+ public boolean excludeAcknowledgement(int messageCount) {
+ return evaluateModulus(messageCount, excludeAcknowledgementModulus);
+ }
+
+ public boolean excludeEndOfBlock(int messageCount) {
+ return evaluateModulus(messageCount, excludeEndOfBlockModulus);
+ }
+
+ public boolean excludeEndOfData(int messageCount) {
+ return evaluateModulus(messageCount, excludeEndOfDataModulus);
+ }
+
+ public boolean disconnectBeforeAcknowledgement(int messageCount) {
+ return evaluateModulus(messageCount, disconnectBeforeAcknowledgementModulus);
+ }
+
+ public boolean disconnectAfterAcknowledgement(int messageCount) {
+ return evaluateModulus(messageCount, disconnectAfterAcknowledgementModulus);
+ }
+
+ public boolean sendOutOfBandData(int messageCount) {
+ return evaluateModulus(messageCount, sendOutOfBandDataModulus);
+ }
+
+ private boolean evaluateModulus(int messageCount, int modulus) {
+ switch (modulus) {
+ case 0:
+ return false;
+ case 1:
+ return true;
+ default:
+ return (messageCount % modulus == 0) ? true : false;
+
+ }
+ }
+
+ private boolean evaluatePatten(String hl7Message, Pattern pattern) {
+ boolean retValue = false;
+
+ if (null != pattern && pattern.matcher(hl7Message).matches()) {
+ retValue = true;
+ }
+
+ return retValue;
+ }
+
+
+ public boolean isActive() {
+ return active;
+ }
+
+ public void setActive(boolean active) {
+ this.active = active;
+ }
+
+ public int getExcludeStartOfBlockModulus() {
+ return excludeStartOfBlockModulus;
+ }
+
+ /**
+ * Set the modulus used to determine when to include the bMLLP_ENVELOPE_START_OF_BLOCK
+ * portion of the MLLP Envelope.
+ * <p/>
+ * If this value is less than or equal to 0, the bMLLP_ENVELOPE_START_OF_BLOCK portion
+ * of the MLLP Envelope will always be included.
+ * If the value is 1, the bMLLP_ENVELOPE_START_OF_BLOCK portion of the MLLP Envelope will
+ * never be included.
+ * Otherwise, if the result of evaluating message count % value is greater
+ * than 0, the bMLLP_ENVELOPE_START_OF_BLOCK portion of the MLLP Envelope will not be
+ * included. Effectively leaving the bMLLP_ENVELOPE_START_OF_BLOCK portion of the MLLP Envelope
+ * out of every n-th message.
+ *
+ * @param excludeStartOfBlockModulus exclude on every n-th message
+ * 0 => Never excluded
+ * 1 => Always excluded
+ */
+ public void setExcludeStartOfBlockModulus(int excludeStartOfBlockModulus) {
+ if (0 > excludeStartOfBlockModulus) {
+ this.excludeStartOfBlockModulus = 0;
+ } else {
+ this.excludeStartOfBlockModulus = excludeStartOfBlockModulus;
+ }
+ }
+
+ public void enableMllpEnvelope() {
+ this.setExcludeStartOfBlockModulus(0);
+ this.setExcludeEndOfBlockModulus(0);
+ this.setExcludeEndOfDataModulus(0);
+ }
+
+ public void disableMllpEnvelopeStart() {
+ this.disableMllpEnvelopeStart(1);
+ }
+
+ public void disableMllpEnvelopeStart(int mllpEnvelopeModulus) {
+ this.setExcludeStartOfBlockModulus(mllpEnvelopeModulus);
+ }
+
+ public void disableMllpEnvelopeEnd() {
+ this.disableMllpEnvelope(1);
+ }
+
+ public void disableMllpEnvelopeEnd(int mllpEnvelopeModulus) {
+ this.setExcludeEndOfBlockModulus(mllpEnvelopeModulus);
+ this.setExcludeEndOfDataModulus(mllpEnvelopeModulus);
+ }
+
+ public void disableMllpEnvelope() {
+ this.disableMllpEnvelope(1);
+ }
+
+ public void disableMllpEnvelope(int mllpEnvelopeModulus) {
+ this.setExcludeStartOfBlockModulus(mllpEnvelopeModulus);
+ this.setExcludeEndOfBlockModulus(mllpEnvelopeModulus);
+ this.setExcludeEndOfDataModulus(mllpEnvelopeModulus);
+ }
+
+ public void enableResponse() {
+ this.setExcludeStartOfBlockModulus(0);
+ this.setExcludeAcknowledgementModulus(0);
+ this.setExcludeEndOfBlockModulus(0);
+ this.setExcludeEndOfDataModulus(0);
+ }
+
+ public void disableResponse() {
+ this.disableResponse(1);
+ }
+
+ public void disableResponse(int mllpResponseModulus) {
+ this.setExcludeStartOfBlockModulus(mllpResponseModulus);
+ this.setExcludeAcknowledgementModulus(mllpResponseModulus);
+ this.setExcludeEndOfBlockModulus(mllpResponseModulus);
+ this.setExcludeEndOfDataModulus(mllpResponseModulus);
+ }
+
+ public int getExcludeEndOfBlockModulus() {
+ return excludeEndOfBlockModulus;
+ }
+
+ public void setExcludeEndOfBlockModulus(int excludeEndOfBlockModulus) {
+ if (0 > excludeEndOfBlockModulus) {
+ this.excludeEndOfBlockModulus = 0;
+ } else {
+ this.excludeEndOfBlockModulus = excludeEndOfBlockModulus;
+ }
+ }
+
+ public int getExcludeEndOfDataModulus() {
+ return excludeEndOfDataModulus;
+ }
+
+ public void setExcludeEndOfDataModulus(int excludeEndOfDataModulus) {
+ if (0 > excludeEndOfDataModulus) {
+ this.excludeEndOfDataModulus = 0;
+ } else {
+ this.excludeEndOfDataModulus = excludeEndOfDataModulus;
+ }
+ }
+
+ public int getExcludeAcknowledgementModulus() {
+ return excludeAcknowledgementModulus;
+ }
+
+ public void setExcludeAcknowledgementModulus(int excludeAcknowledgementModulus) {
+ if (0 > excludeAcknowledgementModulus) {
+ this.excludeAcknowledgementModulus = 0;
+ } else {
+ this.excludeAcknowledgementModulus = excludeAcknowledgementModulus;
+ }
+ }
+
+
+ public int getSendOutOfBandDataModulus() {
+ return sendOutOfBandDataModulus;
+ }
+
+ public void setSendOutOfBandDataModulus(int sendOutOfBandDataModulus) {
+ if (0 > sendOutOfBandDataModulus) {
+ this.sendOutOfBandDataModulus = 0;
+ } else {
+ this.sendOutOfBandDataModulus = sendOutOfBandDataModulus;
+ }
+ }
+
+ public int getDisconnectBeforeAcknowledgementModulus() {
+ return disconnectBeforeAcknowledgementModulus;
+ }
+
+ public void setDisconnectBeforeAcknowledgementModulus(int disconnectBeforeAcknowledgementModulus) {
+ if (0 > disconnectBeforeAcknowledgementModulus) {
+ this.disconnectBeforeAcknowledgementModulus = 0;
+ } else {
+ this.disconnectBeforeAcknowledgementModulus = disconnectBeforeAcknowledgementModulus;
+ }
+ }
+
+ public int getDisconnectAfterAcknowledgementModulus() {
+ return disconnectAfterAcknowledgementModulus;
+ }
+
+ public void setDisconnectAfterAcknowledgementModulus(int disconnectAfterAcknowledgementModulus) {
+ if (0 > disconnectAfterAcknowledgementModulus) {
+ this.disconnectAfterAcknowledgementModulus = 0;
+ } else {
+ this.disconnectAfterAcknowledgementModulus = disconnectAfterAcknowledgementModulus;
+ }
+ }
+
+ public int getSendApplicationRejectAcknowledgementModulus() {
+ return sendApplicationRejectAcknowledgementModulus;
+ }
+
+ public void setSendApplicationRejectAcknowledgementModulus(int sendApplicationRejectAcknowledgementModulus) {
+ if (0 > sendApplicationRejectAcknowledgementModulus) {
+ this.sendApplicationRejectAcknowledgementModulus = 0;
+ } else {
+ this.sendApplicationRejectAcknowledgementModulus = sendApplicationRejectAcknowledgementModulus;
+ }
+ }
+
+ public int getSendApplicationErrorAcknowledgementModulus() {
+ return sendApplicationErrorAcknowledgementModulus;
+ }
+
+ public void setSendApplicationErrorAcknowledgementModulus(int sendApplicationErrorAcknowledgementModulus) {
+ if (0 > sendApplicationErrorAcknowledgementModulus) {
+ this.sendApplicationErrorAcknowledgementModulus = 0;
+ } else {
+ this.sendApplicationErrorAcknowledgementModulus = sendApplicationErrorAcknowledgementModulus;
+ }
+ }
+
+ public Pattern getSendApplicationRejectAcknowledgementPattern() {
+ return sendApplicationRejectAcknowledgementPattern;
+ }
+
+ public void setSendApplicationRejectAcknowledgementPattern(Pattern sendApplicationRejectAcknowledgementPattern) {
+ this.sendApplicationRejectAcknowledgementPattern = sendApplicationRejectAcknowledgementPattern;
+ }
+
+ public Pattern getSendApplicationErrorAcknowledgementPattern() {
+ return sendApplicationErrorAcknowledgementPattern;
+ }
+
+ public void setSendApplicationErrorAcknowledgementPattern(Pattern sendApplicationErrorAcknowledgementPattern) {
+ this.sendApplicationErrorAcknowledgementPattern = sendApplicationErrorAcknowledgementPattern;
+ }
+
+ public ServerSocketThread getServerSocketThread() {
+ return serverSocketThread;
+ }
+
+ public void setServerSocketThread(ServerSocketThread serverSocketThread) {
+ this.serverSocketThread = serverSocketThread;
+ }
+
+ void closeConnection(Socket socket) {
+ if (null != socket) {
+ if (!socket.isClosed()) {
+ try {
+ socket.shutdownInput();
+ } catch (Exception ex) {
+ log.warn("Exception encountered shutting down the input stream on the client socket", ex);
+ }
+
+ try {
+ socket.shutdownOutput();
+ } catch (Exception ex) {
+ log.warn("Exception encountered shutting down the output stream on the client socket", ex);
+ }
+
+ try {
+ socket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing the client socket", ex);
+ }
+ }
+ }
+ }
+
+ void resetConnection(Socket socket) {
+ if (null != socket) {
+ try {
+ socket.setSoLinger(true, 0);
+ } catch (Exception ex) {
+ log.warn("Exception encountered setting SO_LINGER to 0 on the socket to force a reset", ex);
+ } finally {
+ closeConnection(socket);
+ }
+ }
+
+ }
+
+ /**
+ * Nested class to accept TCP connections
+ */
+ class ServerSocketThread extends Thread {
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ ServerSocket serverSocket;
+
+ String listenHost = "0.0.0.0";
+ int listenPort;
+ int backlog = 5;
+
+ int acceptTimeout = 5000;
+
+ boolean raiseExceptionOnAcceptTimeout;
+
+ public ServerSocketThread() throws IOException {
+ bind();
+ }
+
+ public ServerSocketThread(int listenPort) throws IOException {
+ this.listenPort = listenPort;
+ bind();
+ }
+
+ public ServerSocketThread(int listenPort, int backlog) throws IOException {
+ this.listenPort = listenPort;
+ this.backlog = backlog;
+ bind();
+ }
+
+ public ServerSocketThread(String listenHost, int listenPort, int backlog) throws IOException {
+ this.listenHost = listenHost;
+ this.listenPort = listenPort;
+ this.backlog = backlog;
+ bind();
+ }
+
+ /**
+ * Open the TCP Listener
+ *
+ * @throws IOException
+ */
+ private void bind() throws IOException {
+ this.setDaemon(true);
+ serverSocket = new ServerSocket();
+
+ // Set TCP Parameters
+ serverSocket.setSoTimeout(acceptTimeout);
+ serverSocket.setReuseAddress(false);
+
+ if (0 >= listenPort) {
+ serverSocket.bind(null, backlog);
+ } else {
+ serverSocket.bind(new InetSocketAddress(this.listenHost, this.listenPort), backlog);
+ }
+
+ if (0 >= this.listenPort) {
+ this.listenPort = serverSocket.getLocalPort();
+ }
+
+
+ log.info("Opened TCP Listener on port {}", serverSocket.getLocalPort());
+ }
+
+ /**
+ * Accept TCP connections and create ClientSocketThreads for them
+ */
+ public void run() {
+ log.info("Accepting connections on port {}", serverSocket.getLocalPort());
+ this.setName("MllpServerResource$ServerSocketThread - " + serverSocket.getLocalSocketAddress().toString());
+ while (isActive() && serverSocket.isBound()) {
+ Socket clientSocket = null;
+ try {
+ clientSocket = serverSocket.accept();
+ clientSocket.setKeepAlive(true);
+ clientSocket.setTcpNoDelay(false);
+ clientSocket.setSoLinger(false, -1);
+ clientSocket.setSoTimeout(5000);
+ ClientSocketThread clientSocketThread = new ClientSocketThread(clientSocket);
+ clientSocketThread.setDaemon(true);
+ clientSocketThread.start();
+ } catch (SocketTimeoutException timeoutEx) {
+ if (raiseExceptionOnAcceptTimeout) {
+ throw new MllpJUnitResourceTimeoutException("Timeout Accepting client connection", timeoutEx);
+ }
+ continue;
+ } catch (IOException e) {
+ log.warn("IOException creating Client Socket");
+ try {
+ clientSocket.close();
+ } catch (IOException e1) {
+ log.warn("Exceptiong encountered closing client socket after attempting to accept connection");
+ }
+ throw new MllpJUnitResourceException("IOException creating Socket", e);
+ }
+ }
+ log.info("No longer accepting connections - closing TCP Listener on port {}", serverSocket.getLocalPort());
+ try {
+ serverSocket.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ log.info("Closed TCP Listener on port {}", serverSocket.getLocalPort());
+ }
+
+ public void shutdown() {
+ this.interrupt();
+ }
+
+ public String getListenHost() {
+ return listenHost;
+ }
+
+ public int getListenPort() {
+ return listenPort;
+ }
+
+ public int getBacklog() {
+ return backlog;
+ }
+
+ public int getAcceptTimeout() {
+ return acceptTimeout;
+ }
+
+ /**
+ * Enable/disable a timeout while waiting for a TCP connection, in milliseconds. With this option set to a
+ * non-zero timeout, the ServerSocketThread will block for only this amount of time while waiting for a tcp
+ * connection. If the timeout expires and raiseExceptionOnAcceptTimeout is set to true, a MllpJUnitResourceTimeoutException
+ * is raised. Otherwise, the ServerSocketThread will continue to poll for new TCP connections.
+ *
+ * @param acceptTimeout the timeout in milliseconds - zero is interpreted as an infinite timeout
+ */
+ public void setAcceptTimeout(int acceptTimeout) {
+ this.acceptTimeout = acceptTimeout;
+ }
+
+ public boolean isRaiseExceptionOnAcceptTimeout() {
+ return raiseExceptionOnAcceptTimeout;
+ }
+
+ /**
+ * Enable/Disable the generation of MllpJUnitResourceTimeoutException if the ServerSocket.accept()
+ * call raises a SocketTimeoutException.
+ *
+ * @param raiseExceptionOnAcceptTimeout true enables exceptions on an accept timeout
+ */
+ public void setRaiseExceptionOnAcceptTimeout(boolean raiseExceptionOnAcceptTimeout) {
+ this.raiseExceptionOnAcceptTimeout = raiseExceptionOnAcceptTimeout;
+ }
+ }
+
+ /**
+ * Nested class that handles the established TCP connections
+ */
+ class ClientSocketThread extends Thread {
+ /*
+ final char cCARRIAGE_RETURN = 13;
+ final char cLINE_FEED = 10;
+ final char cSEGMENT_DELIMITER = cCARRIAGE_RETURN;
+ final String sMESSAGE_TERMINATOR = "" + cCARRIAGE_RETURN + cLINE_FEED;
+ final byte bMLLP_ENVELOPE_START_OF_BLOCK = 0x0b;
+ final byte bMLLP_ENVELOPE_END_OF_BLOCK = 0x1c;
+ final byte bMLLP_ENVELOPE_END_OF_DATA = 0x0d;
+ final int iEND_OF_TRANSMISSION = -1;
+ */
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ Socket clientSocket;
+
+ int messageCounter;
+
+ ClientSocketThread(Socket clientSocket) {
+ this.clientSocket = clientSocket;
+ }
+
+ /**
+ * Receives HL7 messages and replies with HL7 Acknowledgements.
+ *
+ * The exact behaviour of this method is very configurable, allowing simulation of varies
+ * error conditions.
+ */
+ public void run() {
+ String localAddress = clientSocket.getLocalAddress().toString();
+ String remoteAddress = clientSocket.getRemoteSocketAddress().toString();
+
+ log.info("Handling Connection: {} -> {}", localAddress, remoteAddress);
+
+ try {
+ while (null != clientSocket && clientSocket.isConnected() && !clientSocket.isClosed()) {
+ InputStream instream = clientSocket.getInputStream();
+ String parsedHL7Message = getMessage(instream);
+
+ if (null != parsedHL7Message && parsedHL7Message.length() > 0) {
+ ++messageCounter;
+ if (disconnectBeforeAcknowledgement(messageCounter)) {
+ log.warn("Disconnecting before sending acknowledgement");
+ clientSocket.shutdownInput();
+ clientSocket.shutdownOutput();
+ clientSocket.close();
+ break;
+ }
+
+ String acknowledgmentMessage;
+
+ if (sendApplicationErrorAcknowledgement(messageCounter) || sendApplicationErrorAcknowledgement(parsedHL7Message)) {
+ acknowledgmentMessage = generateAcknowledgementMessage(parsedHL7Message, "AE");
+ } else if (sendApplicationRejectAcknowledgement(messageCounter) || sendApplicationRejectAcknowledgement(parsedHL7Message)) {
+ acknowledgmentMessage = generateAcknowledgementMessage(parsedHL7Message, "AR");
+ } else {
+ acknowledgmentMessage = generateAcknowledgementMessage(parsedHL7Message);
+
+ }
+ BufferedOutputStream outstream = new BufferedOutputStream(clientSocket.getOutputStream());
+
+ if (sendOutOfBandData(messageCounter)) {
+ byte[] outOfBandDataBytes = "Out Of Band Hl7MessageGenerator".getBytes();
+ outstream.write(outOfBandDataBytes, 0, outOfBandDataBytes.length);
+
+ }
+ if (excludeStartOfBlock(messageCounter)) {
+ log.warn("NOT sending bMLLP_ENVELOPE_START_OF_BLOCK");
+ } else {
+ outstream.write(START_OF_BLOCK);
+ }
+
+ if (excludeAcknowledgement(messageCounter)) {
+ log.info("NOT sending Acknowledgement body");
+ } else {
+ log.debug("Buffering Acknowledgement\n\t{}", acknowledgmentMessage.replace('\r', '\n'));
+ byte[] ackBytes = acknowledgmentMessage.getBytes();
+ outstream.write(ackBytes, 0, ackBytes.length);
+ }
+
+ if (excludeEndOfBlock(messageCounter)) {
+ log.warn("NOT sending bMLLP_ENVELOPE_END_OF_BLOCK");
+ } else {
+ outstream.write(END_OF_BLOCK);
+ }
+
+ if (excludeEndOfData(messageCounter)) {
+ log.warn("NOT sending bMLLP_ENVELOPE_END_OF_DATA");
+ } else {
+ outstream.write(END_OF_DATA);
+ }
+
+ log.debug("Writing Acknowledgement\n\t{}", acknowledgmentMessage.replace('\r', '\n'));
+ outstream.flush();
+
+ if (disconnectAfterAcknowledgement(messageCounter)) {
+ log.info("Closing Client");
+ clientSocket.shutdownInput();
+ clientSocket.shutdownOutput();
+ clientSocket.close();
+ break;
+ }
+ }
+ }
+ } catch (IOException e) {
+ String errorMessage = "Error whiling reading and writing to clientSocket";
+ log.error(errorMessage, e);
+ throw new MllpJUnitResourceException(errorMessage, e);
+ } finally {
+ try {
+ clientSocket.close();
+ } catch (IOException e) {
+ String errorMessage = "Error whiling attempting to close to client Socket";
+ log.error(errorMessage, e);
+ throw new MllpJUnitResourceException(errorMessage, e);
+ }
+ }
+
+ log.info("Connection Finished: {} -> {}", localAddress, remoteAddress);
+ }
+
+ /**
+ * Read a MLLP-Framed message
+ *
+ * @param anInputStream source input stream
+ * @return the MLLP payload
+ * @throws IOException when the underlying Java Socket calls raise these exceptions
+ */
+ // TODO: Enhance this to detect non-HL7 data (i.e. look for MSH after START_OF_BLOCK)
+ public String getMessage(InputStream anInputStream) throws IOException {
+ try {
+ // TODO: Enhance this to read a bunch of characters and log, rather than log them one at a time
+ boolean waitingForStartOfBlock = true;
+ while (waitingForStartOfBlock) {
+ int potentialStartCharacter = anInputStream.read();
+ switch (potentialStartCharacter) {
+ case END_OF_STREAM:
+ return null;
+ case START_OF_BLOCK:
+ waitingForStartOfBlock = false;
+ break;
+ default:
+ log.warn("START_OF_BLOCK character has not been received. Out-of-band character received: {}", potentialStartCharacter);
+ }
+ }
+ } catch (SocketException socketEx) {
+ log.error("Unable to read from socket stream when expected bMLLP_ENVELOPE_START_OF_BLOCK - resetting connection ", socketEx);
+ resetConnection(clientSocket);
+ return null;
+ }
+
+ boolean endOfMessage = false;
+ StringBuilder parsedMessage = new StringBuilder(anInputStream.available() + 10);
+ while (!endOfMessage) {
+ int characterReceived = anInputStream.read();
+
+ switch (characterReceived) {
+ case START_OF_BLOCK:
+ log.error("Received START_OF_BLOCK before END_OF_DATA. Discarding data: {}", parsedMessage.toString());
+ return null;
+ case END_OF_STREAM:
+ log.error("Received END_OF_STREAM without END_OF_DATA. Discarding data: {}", parsedMessage.toString());
+ return null;
+ case END_OF_BLOCK:
+ characterReceived = anInputStream.read();
+ if (characterReceived != END_OF_DATA) {
+ log.error("Received {} when expecting END_OF_DATA after END_OF_BLOCK. Discarding Hl7MessageGenerator: {}",
+ characterReceived, parsedMessage.toString());
+ return null;
+ }
+ endOfMessage = true;
+ break;
+ default:
+ parsedMessage.append((char) characterReceived);
+ }
+
+ }
+
+ return parsedMessage.toString();
+ }
+
+ /**
+ * Generates a HL7 Application Accept Acknowledgement
+ *
+ * @param hl7Message HL7 message that is being acknowledged
+ * @return a HL7 Application Accept Acknowlegdement
+ */
+ private String generateAcknowledgementMessage(String hl7Message) {
+ return generateAcknowledgementMessage(hl7Message, "AA");
+ }
+
+ /**
+ * Generates a HL7 Application Acknowledgement
+ *
+ * @param hl7Message HL7 message that is being acknowledged
+ * @param acknowledgementCode AA, AE or AR
+ * @return a HL7 Application Acknowledgement
+ */
+ private String generateAcknowledgementMessage(String hl7Message, String acknowledgementCode) {
+ final String defaulNackMessage =
+ "MSH|^~\\&|||||||NACK||P|2.2" + SEGMENT_DELIMITER
+ + "MSA|AR|" + SEGMENT_DELIMITER
+ + MESSAGE_TERMINATOR;
+
+ if (hl7Message == null) {
+ log.error("Invalid HL7 message for parsing operation. Please check your inputs");
+ return defaulNackMessage;
+ }
+
+ if (!("AA".equals(acknowledgementCode) || "AE".equals(acknowledgementCode) || "AR".equals(acknowledgementCode))) {
+ throw new IllegalArgumentException("Acknowledgemnt Code must be AA, AE or AR: " + acknowledgementCode);
+ }
+
+ String messageControlId;
+
+ int endOfMshSegment = hl7Message.indexOf(SEGMENT_DELIMITER);
+ if (-1 != endOfMshSegment) {
+ String mshSegment = hl7Message.substring(0, endOfMshSegment);
+ char fieldSeparator = mshSegment.charAt(3);
+ String fieldSeparatorPattern = Pattern.quote("" + fieldSeparator);
+ String[] mshFields = mshSegment.split(fieldSeparatorPattern);
+ if (null == mshFields || 0 == mshFields.length) {
+ log.error("Failed to split MSH Segment into fields");
+ } else {
+ StringBuilder ackBuilder = new StringBuilder(mshSegment.length() + 25);
+ // Build the MSH Segment
+ ackBuilder
+ .append(mshFields[0]).append(fieldSeparator)
+ .append(mshFields[1]).append(fieldSeparator)
+ .append(mshFields[4]).append(fieldSeparator)
+ .append(mshFields[5]).append(fieldSeparator)
+ .append(mshFields[2]).append(fieldSeparator)
+ .append(mshFields[3]).append(fieldSeparator)
+ .append(mshFields[6]).append(fieldSeparator)
+ .append(mshFields[7]).append(fieldSeparator)
+ .append("ACK")
+ .append(mshFields[8].substring(3));
+ for (int i = 9; i < mshFields.length; ++i) {
+ ackBuilder.append(fieldSeparator).append(mshFields[i]);
+ }
+ // Empty fields at the end are not preserved by String.split, so preserve them
+ int emptyFieldIndex = mshSegment.length() - 1;
+ if (fieldSeparator == mshSegment.charAt(mshSegment.length() - 1)) {
+ ackBuilder.append(fieldSeparator);
+ while (emptyFieldIndex >= 1 && mshSegment.charAt(emptyFieldIndex) == mshSegment.charAt(emptyFieldIndex - 1)) {
+ ackBuilder.append(fieldSeparator);
+ --emptyFieldIndex;
+ }
+ }
+ ackBuilder.append(SEGMENT_DELIMITER);
+
+ // Build the MSA Segment
+ ackBuilder
+ .append("MSA").append(fieldSeparator)
+ .append(acknowledgementCode).append(fieldSeparator)
+ .append(mshFields[9]).append(fieldSeparator)
+ .append(SEGMENT_DELIMITER);
+
+ // Terminate the message
+ ackBuilder.append(MESSAGE_TERMINATOR);
+
+ return ackBuilder.toString();
+ }
+ } else {
+ log.error("Failed to find the end of the MSH Segment");
+ }
+
+ return null;
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/Hl7MessageGenerator.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/Hl7MessageGenerator.java b/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/Hl7MessageGenerator.java
new file mode 100644
index 0000000..0b12037
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/Hl7MessageGenerator.java
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.mllp;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+public final class Hl7MessageGenerator {
+ static SimpleDateFormat timestampFormat = new SimpleDateFormat("YYYYMMddHHmmss");
+ static String messageControlIdFormat = "%05d";
+
+ static String hl7MessageTemplate =
+ "MSH|^~\\&|ADT|EPIC|JCAPS|CC|<MESSAGE_TIMESTAMP>|RISTECH|ADT^A08|<MESSAGE_CONTROL_ID>|D|2.3^^|||||||" + '\r'
+ + "EVN|A08|20150107161440||REG_UPDATE_SEND_VISIT_MESSAGES_ON_PATIENT_CHANGES|RISTECH^RADIOLOGY^TECHNOLOGIST^^^^^^UCLA^^^^^RRMC||" + '\r'
+ + "PID|1|2100355^^^MRN^MRN|2100355^^^MRN^MRN||MDCLS9^MC9||19700109|F||U|111 HOVER STREET^^LOS ANGELES^CA^90032^USA^P^^LOS ANGELE|LOS ANGELE|(310)"
+ + "725-6952^P^PH^^^310^7256952||ENGLISH|U||60000013647|565-33-2222|||U||||||||N||" + '\r'
+ + "PD1|||UCLA HEALTH SYSTEM^^10|10002116^ADAMS^JOHN^D^^^^^EPIC^^^^PROVID||||||||||||||" + '\r'
+ + "NK1|1|DOE^MC9^^|OTH|^^^^^USA|(310)888-9999^^^^^310^8889999|(310)999-2222^^^^^310^9992222|Emergency Contact 1|||||||||||||||||||||||||||" + '\r'
+ + "PV1|1|OUTPATIENT|RR CT^^^1000^^^^^^^DEPID|EL|||017511^TOBIAS^JONATHAN^^^^^^EPIC^^^^PROVID|017511^TOBIAS^JONATHAN^^^^^^EPIC^^^^PROVID||||||CLR|||||60000013647|SELF"
+ + "|||||||||||||||||||||HOV_CONF|^^^1000^^^^^^^||20150107161438||||||||||" + '\r'
+ + "PV2||||||||20150107161438||||CT BRAIN W WO CONTRAST||||||||||N|||||||||||||||||||||||||||" + '\r'
+ + "ZPV||||||||||||20150107161438|||||||||" + '\r'
+ + "AL1|1||33361^NO KNOWN ALLERGIES^^NOTCOMPUTRITION^NO KNOWN ALLERGIES^EXTELG||||||" + '\r'
+ + "DG1|1|DX|784.0^Headache^DX|Headache||VISIT" + '\r'
+ + "GT1|1|1000235129|MDCLS9^MC9^^||111 HOVER STREET^^LOS ANGELES^CA^90032^USA^^^LOS ANGELE|(310)"
+ + "725-6952^^^^^310^7256952||19700109|F|P/F|SLF|565-33-2222|||||^^^^^USA|||UNKNOWN|||||||||||||||||||||||||||||" + '\r'
+ + "UB2||||||||" + '\r'
+ + '\r' + '\n';
+ private Hl7MessageGenerator() {
+ }
+
+ public static String generateMessage() {
+ return generateMessage(new Date(), 1);
+ }
+
+ public static String generateMessage(int messageControlId) {
+ return generateMessage(new Date(), messageControlId);
+ }
+
+ public static String generateMessage(Date timestamp, int messageControlId) {
+ String tmpMessage = hl7MessageTemplate.replaceFirst("<MESSAGE_TIMESTAMP>", timestampFormat.format(timestamp));
+ return tmpMessage.replaceFirst("<MESSAGE_CONTROL_ID>", String.format("%05d", messageControlId));
+ }
+
+ public static String getHl7MessageTemplate() {
+ return hl7MessageTemplate;
+ }
+
+ public static void setHl7MessageTemplate(String hl7MessageTemplate) {
+ Hl7MessageGenerator.hl7MessageTemplate = hl7MessageTemplate;
+ }
+
+ public static SimpleDateFormat getTimestampFormat() {
+ return timestampFormat;
+ }
+
+ public static void setTimestampFormat(SimpleDateFormat timestampFormat) {
+ Hl7MessageGenerator.timestampFormat = timestampFormat;
+ }
+
+ public static String getMessageControlIdFormat() {
+ return messageControlIdFormat;
+ }
+
+ public static void setMessageControlIdFormat(String messageControlIdFormat) {
+ Hl7MessageGenerator.messageControlIdFormat = messageControlIdFormat;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/PassthroughProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/PassthroughProcessor.java b/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/PassthroughProcessor.java
new file mode 100644
index 0000000..eb054df
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/mllp/PassthroughProcessor.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.mllp;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A Passthrough test processor
+ *
+ * Used to set breakpoints and diagnose what is going on in test routes
+ */
+public class PassthroughProcessor implements Processor {
+ String id;
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ public PassthroughProcessor(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String message = exchange.getIn().getBody(String.class);
+ if (null != message) {
+ String msh = message.substring(0, message.indexOf('\r'));
+ log.debug("Processing MSH {}: \n{}\n", id, msh);
+ }
+
+ log.debug("Null inbound message body");
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/tcp/JavaSocketTests.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/tcp/JavaSocketTests.java b/components/camel-mllp/src/test/java/org/apache/camel/test/tcp/JavaSocketTests.java
new file mode 100644
index 0000000..be0b44f
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/tcp/JavaSocketTests.java
@@ -0,0 +1,176 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.tcp;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Various tests used to validate the behaviour of Java Sockets.
+ *
+ * The tests were for experimentation and don't have any assertions in them - JUnit
+ * provided a convenient framework to explore this behaviour. These tests shouldn't be
+ * run with a normal build since they don't have any assertions and don't validate any
+ * results.
+ *
+ * NOTE: This class may be deleted in the future
+ */
+@Ignore(value = "Tests validating Java Socket behaviours")
+public class JavaSocketTests {
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ Socket clientSocket;
+ ServerSocket serverSocket;
+
+ int messageCount = 10;
+
+ @Before
+ public void setUp() throws Exception {
+ serverSocket = new ServerSocket(0);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (null != clientSocket) {
+ clientSocket.close();
+ }
+
+ if (null != serverSocket) {
+ serverSocket.close();
+ }
+ }
+
+ @Test
+ public void testSocketReadOnClosedConnection() throws Exception {
+ final Thread acceptThread = new Thread() {
+ Logger log = LoggerFactory.getLogger("acceptThread");
+
+ @Override
+ public void run() {
+ boolean running = true;
+ try {
+ Socket echoSocket = serverSocket.accept();
+
+ log.info("Accepted connection: {}", echoSocket.getInetAddress());
+
+ echoSocket.setSoTimeout(2000);
+
+ while (echoSocket.isConnected() && !echoSocket.isClosed()) {
+ StringBuilder responseBuilder = new StringBuilder(500);
+ InputStream reader = echoSocket.getInputStream();
+ OutputStream writer = echoSocket.getOutputStream();
+
+ do {
+ int readByte = -1;
+ int available = -1;
+ try {
+ available = reader.available();
+ log.info("InputStream.available returned {}", available);
+ readByte = reader.read();
+ log.trace("Processing byte: {}", readByte);
+ switch (readByte) {
+ case -1:
+ if (echoSocket.isConnected() && !echoSocket.isClosed()) {
+ log.info("Available returned {}", reader.available());
+ log.warn("Socket claims to still be open, but END_OF_STREAM received - closing echoSocket");
+ try {
+ echoSocket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing echoSocket after END_OF_STREAM received", ex);
+ }
+ }
+ running = false;
+ break;
+ case 10:
+ log.info("Complete Message - Sending Response");
+ byte[] response = responseBuilder.toString().getBytes();
+ responseBuilder.setLength(0);
+ writer.write(response, 0, response.length);
+ writer.write('\n');
+ break;
+ default:
+ responseBuilder.append((char) readByte);
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ log.info("Timeout reading data - available returned {}", available);
+ }
+ } while (echoSocket.isConnected() && !echoSocket.isClosed());
+ }
+
+ } catch (IOException ioEx) {
+ log.error("IOException in run method", ioEx);
+ } finally {
+ try {
+ serverSocket.close();
+ } catch (IOException ioEx) {
+ log.error("Exception encountered closing server socket", ioEx);
+ }
+ }
+
+
+ log.info("Finished processing connection");
+ }
+
+ };
+
+ acceptThread.start();
+
+ clientSocket = new Socket();
+ clientSocket.setSoTimeout(1000);
+ clientSocket.connect(serverSocket.getLocalSocketAddress(), 10000);
+ clientSocket.setTcpNoDelay(true);
+ log.info("Begining message send loop ");
+ byte[] message = "Hello World".getBytes();
+ BufferedReader reader;
+ for (int i = 1; i <= messageCount; ++i) {
+ reader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+ OutputStream writer = clientSocket.getOutputStream();
+ log.info("Sending payload");
+ writer.write(message, 0, message.length);
+ writer.flush();
+ log.info("Sending terminator");
+ writer.write('\n');
+ writer.flush();
+ log.info("Received Response #{}: {}", i, reader.readLine());
+ Thread.sleep(1000);
+ }
+
+ log.info("Message send loop complete - closing connection");
+ // Javadoc for Socket says closing the InputStream will close the connection
+ clientSocket.getInputStream().close();
+ if (!clientSocket.isClosed()) {
+ log.warn("Closing input stream didn't close socket");
+ clientSocket.close();
+ }
+ log.info("Sleeping ...");
+ Thread.sleep(5000);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-client-producer-test.xml
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-client-producer-test.xml b/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-client-producer-test.xml
new file mode 100644
index 0000000..627ed03
--- /dev/null
+++ b/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-client-producer-test.xml
@@ -0,0 +1,114 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd
+ http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.1.0.xsd
+ http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd">
+
+ <property-placeholder persistent-id="MllpTcpClientProducer" update-strategy="reload" xmlns="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0" >
+ <default-properties>
+ <property name="sourceUri" value="direct://source" />
+ <property name="acknowledgedUri" value="mock://acknowledged" />
+ <property name="timeoutUri" value="mock://timeout-ex" />
+ <property name="frameErrorUri" value="mock://frame-ex" />
+ <property name="errorAcknowledgementUri" value="mock://ae-ack" />
+ <property name="rejectAcknowledgementUri" value="mock://ar-ack" />
+ <property name="mllp.port" value="7777" />
+ </default-properties>
+ </property-placeholder>
+
+
+ <errorHandler id="DefaultErrorHandler" type="DefaultErrorHandler" xmlns="http://camel.apache.org/schema/blueprint">
+ <redeliveryPolicy allowRedeliveryWhileStopping="false"/>
+ </errorHandler>
+
+ <camelContext id="mllp-tcp-client-producer-test" xmlns="http://camel.apache.org/schema/blueprint"
+ useMDCLogging="true" trace="false"
+ >
+
+ <onException>
+ <exception>java.net.ConnectException</exception>
+ <redeliveryPolicy
+ allowRedeliveryWhileStopping="false"
+ redeliveryDelay="500"
+ maximumRedeliveryDelay="30000"
+ backOffMultiplier="2"
+ useExponentialBackOff="true"
+ maximumRedeliveries="-1"
+ logRetryAttempted="true"
+ retryAttemptedLogLevel="WARN"
+ />
+ </onException>
+
+ <onException>
+ <exception>org.apache.camel.component.mllp.MllpCorruptFrameException</exception>
+ <redeliveryPolicy logHandled="true"/>
+ <handled>
+ <constant>true</constant>
+ </handled>
+ <to uri="{{frameErrorUri}}"/>
+ </onException>
+
+ <onException>
+ <exception>org.apache.camel.component.mllp.MllpTimeoutException</exception>
+ <redeliveryPolicy logHandled="true"/>
+ <handled>
+ <constant>true</constant>
+ </handled>
+ <to uri="{{timeoutUri}}"/>
+ </onException>
+
+ <onException>
+ <exception>org.apache.camel.component.mllp.MllpApplicationErrorAcknowledgementException</exception>
+ <redeliveryPolicy logHandled="true" logRetryAttempted="true" retryAttemptedLogLevel="WARN" logExhausted="true" retriesExhaustedLogLevel="ERROR"
+ allowRedeliveryWhileStopping="false"
+ maximumRedeliveries="10"
+ redeliveryDelay="1000"
+ useExponentialBackOff="true"
+ maximumRedeliveryDelay="15000"
+ />
+ <handled>
+ <constant>true</constant>
+ </handled>
+ <to uri="{{errorAcknowledgementUri}}" />
+ </onException>
+
+ <onException>
+ <exception>org.apache.camel.component.mllp.MllpApplicationRejectAcknowledgementException</exception>
+ <redeliveryPolicy logHandled="true" />
+ <handled>
+ <constant>true</constant>
+ </handled>
+ <to uri="{{rejectAcknowledgementUri}}" />
+ </onException>
+
+ <route id="direct-to-mllp">
+ <from uri="{{sourceUri}}" />
+ <log loggingLevel="DEBUG" message="sending message: $simple{header[CamelMllpMessageControlId]}" />
+ <to uri="mllp://{{mllp.port}}"/>
+ <to uri="{{acknowledgedUri}}"/>
+ <log loggingLevel="DEBUG" message="message sent: $simple{header[CamelMllpMessageControlId]}" />
+ <to uri="log:direct-to-mllp?level=INFO&groupSize=50" />
+ </route>
+ </camelContext>
+
+</blueprint>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-server-consumer-test.xml
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-server-consumer-test.xml b/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-server-consumer-test.xml
new file mode 100644
index 0000000..f90ef63
--- /dev/null
+++ b/components/camel-mllp/src/test/resources/OSGI-INF/blueprint/mllp-tcp-server-consumer-test.xml
@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="
+ http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd
+ http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.1.0.xsd
+ http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd">
+
+ <property-placeholder persistent-id="MllpTcpServerConsumerBlueprintTest" update-strategy="reload" xmlns="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0" >
+ <default-properties>
+ <property name="receivedUri" value="mock://received" />
+ <property name="mllp.port" value="7777" />
+ </default-properties>
+ </property-placeholder>
+
+
+
+ <camelContext id="tcp-server-consumer-context" useMDCLogging="true" xmlns="http://camel.apache.org/schema/blueprint">
+ <!-- TODO: Add exception handlers -->
+ <route>
+ <from uri="mllp://{{mllp.port}}" />
+ <to uri="{{receivedUri}}" />
+ </route>
+ </camelContext>
+
+</blueprint>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/resources/log4j.properties b/components/camel-mllp/src/test/resources/log4j.properties
new file mode 100644
index 0000000..29932b4
--- /dev/null
+++ b/components/camel-mllp/src/test/resources/log4j.properties
@@ -0,0 +1,41 @@
+###
+ # Licensed to the Apache Software Foundation (ASF) under one or more
+ # contributor license agreements. See the NOTICE file distributed with
+ # this work for additional information regarding copyright ownership.
+ # The ASF licenses this file to You under the Apache License, Version 2.0
+ # (the "License"); you may not use this file except in compliance with
+ # the License. You may obtain a copy of the License at
+ #
+ # http://www.apache.org/licenses/LICENSE-2.0
+ #
+ # Unless required by applicable law or agreed to in writing, software
+ # distributed under the License is distributed on an "AS IS" BASIS,
+ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ # See the License for the specific language governing permissions and
+ # limitations under the License.
+ ##
+
+log4j.rootLogger=INFO, console, file
+
+# CONSOLE appender
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Target=System.err
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX}] %p | %t | %c %n\t%m%n
+
+# File appender - not normally used in testing
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d %-5p %c{1} - %m %n
+log4j.appender.file.file=target/camel-mllp-test.log
+
+# log4j.logger.org.apache.camel.component.hl7=TRACE
+log4j.logger.org.apache.camel.component.mllp=INFO
+log4j.logger.org.apache.camel.test.mllp.PassthroughProcessor=INFO
+log4j.logger.org.apache.camel=WARN
+log4j.logger.org.apache.activemq=WARN
+log4j.logger.org.apache.camel.test.junit=WARN
+log4j.logger.org.apache.camel.test.tcp=WARN
+
+
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/pom.xml
----------------------------------------------------------------------
diff --git a/components/pom.xml b/components/pom.xml
index a68a15e..8529c15 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -163,6 +163,7 @@
<module>camel-metrics</module>
<module>camel-mina</module>
<module>camel-mina2</module>
+ <module>camel-mllp</module>
<module>camel-mongodb</module>
<module>camel-mqtt</module>
<module>camel-msv</module>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index a7f0751..3c4cdbc 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1230,6 +1230,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-mllp</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-mongodb</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/platforms/karaf/features/src/main/resources/features.xml
----------------------------------------------------------------------
diff --git a/platforms/karaf/features/src/main/resources/features.xml b/platforms/karaf/features/src/main/resources/features.xml
index 178180c..358e1a8 100644
--- a/platforms/karaf/features/src/main/resources/features.xml
+++ b/platforms/karaf/features/src/main/resources/features.xml
@@ -1030,6 +1030,9 @@
<bundle dependency='true'>mvn:org.apache.mina/mina-core/${mina2-version}</bundle>
<bundle>mvn:org.apache.camel/camel-mina2/${project.version}</bundle>
</feature>
+ <feature name='camel-mllp' version='${project.version}' resolver='(obr)' start-level='50'>
+ <bundle>mvn:org.apache.camel/camel-mllp/${project.version}</bundle>
+ </feature>
<feature name='camel-mongodb' version='${project.version}' resolver='(obr)' start-level='50'>
<bundle dependency='true'>mvn:org.mongodb/mongo-java-driver/${mongo-java-driver-version}</bundle>
<bundle>mvn:org.apache.camel/camel-mongodb/${project.version}</bundle>
[5/6] camel git commit: CAMEL-9416: Polished. This closes #748
Posted by da...@apache.org.
CAMEL-9416: Polished. This closes #748
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e5725284
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e5725284
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e5725284
Branch: refs/heads/master
Commit: e5725284485e91093847b5b261171c96ca0aed03
Parents: bd1661b
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Dec 30 09:59:34 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Dec 30 09:59:34 2015 +0100
----------------------------------------------------------------------
components/camel-mllp/pom.xml | 120 +++++------
.../camel/component/mllp/MllpComponent.java | 4 -
.../camel/component/mllp/MllpEndpoint.java | 25 +--
.../component/mllp/MllpTcpClientProducer.java | 3 +-
.../component/mllp/MllpTcpServerConsumer.java | 6 -
.../camel/component/mllp/impl/MllpUtil.java | 80 ++++----
.../mllp/Hl7AcknowledgementGenerator.java | 4 +-
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 203 +++++++++++++++++++
.../src/test/resources/log4j.properties | 69 +++----
10 files changed, 548 insertions(+), 169 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-mllp/pom.xml b/components/camel-mllp/pom.xml
index 3a41bcf..e49249b 100644
--- a/components/camel-mllp/pom.xml
+++ b/components/camel-mllp/pom.xml
@@ -19,78 +19,66 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.camel</groupId>
- <artifactId>components</artifactId>
- <version>2.17-SNAPSHOT</version>
- </parent>
-
- <artifactId>camel-mllp</artifactId>
- <packaging>bundle</packaging>
- <name>Camel :: MLLP</name>
- <description>Camel MLLP Component</description>
-
- <properties>
- <camel.osgi.export.pkg>org.apache.camel.component.mllp.*,org.apache.camel.processor.mllp.*</camel.osgi.export.pkg>
- <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=mllp</camel.osgi.export.service>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-core</artifactId>
- </dependency>
-
-
- <!-- support camel documentation -->
- <!--dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>apt</artifactId>
- </dependency -->
-
- <!-- logging -->
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
-
- <!-- testing -->
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-test</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-test-blueprint</artifactId>
- <scope>test</scope>
- </dependency>
-
- <!-- test logging -->
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.17-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-mllp</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: MLLP</name>
+ <description>Camel MLLP Component</description>
+
+ <properties>
+ <camel.osgi.export.pkg>
+ org.apache.camel.component.mllp.*,
+ org.apache.camel.processor.mllp.*
+ </camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=mllp</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+ <!-- testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-blueprint</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- test logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
<!-- Ensure that test runs do no leave running JVMs -->
- <build>
+ <build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
- <version>2.19</version>
<configuration>
<shutdown>kill</shutdown>
</configuration>
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
index d815158..7a3fbb1 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
@@ -21,8 +21,6 @@ import java.util.Map;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.UriEndpointComponent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Represents the component that manages {@link MllpEndpoint}.
@@ -30,8 +28,6 @@ import org.slf4j.LoggerFactory;
public class MllpComponent extends UriEndpointComponent {
public static final String MLLP_LOG_PHI_PROPERTY = "org.apache.camel.mllp.logPHI";
- Logger log = LoggerFactory.getLogger(this.getClass());
-
public MllpComponent() {
super(MllpEndpoint.class);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
index eda5757..21740a6 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
@@ -45,7 +45,7 @@ public class MllpEndpoint extends DefaultEndpoint {
public static final char SEGMENT_DELIMITER = 0x0d; // CR (carriage return) - decimal 13, octal 015
public static final char MESSAGE_TERMINATOR = 0x0a; // LF (line feed, new line) - decimal 10, octal 012
- Logger log = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(MllpEndpoint.class);
@UriPath(defaultValue = "0.0.0.0", description = "Hostname or IP for connection")
String hostname = "0.0.0.0";
@@ -54,6 +54,7 @@ public class MllpEndpoint extends DefaultEndpoint {
int port = -1;
// TODO: Move URI Params to a MllpConfiguration class
+ // TODO: Move the description documentation to javadoc in the setter method
@UriParam(defaultValue = "5", description = "TCP Server only - The maximum queue length for incoming connection indications (a request to connect) is set to the backlog parameter. If a "
+ "connection indication arrives when the queue is full, the connection is refused.")
int backlog = 5;
@@ -100,6 +101,9 @@ public class MllpEndpoint extends DefaultEndpoint {
public MllpEndpoint(String uri, MllpComponent component) {
super(uri, component);
+ // TODO: this logic should be in component class
+ // TODO: all the options in the endpoint should be getter/setter so you can configure them as plain Java
+
// mllp://hostname:port
String hostPort;
// look for options
@@ -129,15 +133,9 @@ public class MllpEndpoint extends DefaultEndpoint {
}
@Override
- public Exchange createExchange() {
- return this.createExchange(getExchangePattern());
- }
-
- @Override
public Exchange createExchange(ExchangePattern exchangePattern) {
Exchange mllpExchange = super.createExchange(exchangePattern);
setExchangeProperties(mllpExchange);
-
return mllpExchange;
}
@@ -145,26 +143,25 @@ public class MllpEndpoint extends DefaultEndpoint {
public Exchange createExchange(Exchange exchange) {
Exchange mllpExchange = super.createExchange(exchange);
setExchangeProperties(mllpExchange);
-
return mllpExchange;
}
private void setExchangeProperties(Exchange mllpExchange) {
- if (null != charsetName) {
+ if (charsetName != null) {
mllpExchange.setProperty(Exchange.CHARSET_NAME, charsetName);
}
}
public Producer createProducer() throws Exception {
- log.trace("({}).createProducer()", this.getEndpointKey());
-
+ LOG.trace("({}).createProducer()", this.getEndpointKey());
return new MllpTcpClientProducer(this);
}
public Consumer createConsumer(Processor processor) throws Exception {
- log.trace("({}).createConsumer(processor)", this.getEndpointKey());
-
- return new MllpTcpServerConsumer(this, processor);
+ LOG.trace("({}).createConsumer(processor)", this.getEndpointKey());
+ Consumer consumer = new MllpTcpServerConsumer(this, processor);
+ configureConsumer(consumer);
+ return consumer;
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
index 7b9d86b..fdff64b 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
@@ -105,7 +105,7 @@ public class MllpTcpClientProducer extends DefaultProducer {
}
log.debug("Reading acknowledgement from external system");
- byte[] acknowledgementBytes = null;
+ byte[] acknowledgementBytes;
try {
MllpUtil.openFrame(socket);
acknowledgementBytes = MllpUtil.closeFrame(socket);
@@ -226,7 +226,6 @@ public class MllpTcpClientProducer extends DefaultProducer {
}
return null;
-
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
index 362a30f..d74468c 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
@@ -55,10 +55,7 @@ public class MllpTcpServerConsumer extends DefaultConsumer {
public MllpTcpServerConsumer(MllpEndpoint endpoint, Processor processor) {
super(endpoint, processor);
log.trace("MllpTcpServerConsumer(endpoint, processor)");
-
-
this.endpoint = endpoint;
-
}
@Override
@@ -128,7 +125,6 @@ public class MllpTcpServerConsumer extends DefaultConsumer {
super.doShutdown();
}
-
/**
* Nested Class to handle the ServerSocket.accept requests
*/
@@ -543,8 +539,6 @@ public class MllpTcpServerConsumer extends DefaultConsumer {
}
}
}
-
-
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
index 4fd3fbf..c1c5aec 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
@@ -37,7 +37,6 @@ import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_DATA;
import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_STREAM;
import static org.apache.camel.component.mllp.MllpEndpoint.START_OF_BLOCK;
-
/**
* Supplies methods to read and write messages in a MLLP Frame.
* <p/>
@@ -51,7 +50,7 @@ import static org.apache.camel.component.mllp.MllpEndpoint.START_OF_BLOCK;
* <p/>
*/
public final class MllpUtil {
- static Logger log = LoggerFactory.getLogger(MllpUtil.class);
+ private static final Logger LOG = LoggerFactory.getLogger(MllpUtil.class);
private MllpUtil() {
}
@@ -88,7 +87,7 @@ public final class MllpUtil {
// Just pass this on - the caller will wrap it in a MllpTimeoutException
throw normaTimeoutEx;
} catch (IOException unexpectedException) {
- log.error("Unexpected Exception occurred opening MLLP frame - resetting the connection");
+ LOG.error("Unexpected Exception occurred opening MLLP frame - resetting the connection");
MllpUtil.resetConnection(socket);
throw new MllpException("Unexpected Exception occurred opening MLLP frame", unexpectedException);
}
@@ -105,21 +104,21 @@ public final class MllpUtil {
readByte = socketInputStream.read();
switch (readByte) {
case END_OF_STREAM:
- if (isLogPHIEnabled(log)) {
- log.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and "
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and "
+ "out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
outOfFrameData.toString().replace('\r', '\n'));
} else {
- log.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ LOG.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
}
resetConnection(socket);
throw new MllpCorruptFrameException("END_OF_STREAM read while looking for the beginning of the MLLP frame", outOfFrameData.toByteArray());
case START_OF_BLOCK:
- if (isLogPHIEnabled(log)) {
- log.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data: {}", outOfFrameData.toString().replace('\r', '\n'));
+ if (isLogPHIEnabled(LOG)) {
+ LOG.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data: {}", outOfFrameData.toString().replace('\r', '\n'));
} else {
- log.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data");
+ LOG.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data");
}
throw new MllpCorruptFrameException("The beginning of the MLLP frame was preceded by out-of-frame data", outOfFrameData.toByteArray());
@@ -130,22 +129,22 @@ public final class MllpUtil {
}
}
} catch (SocketTimeoutException timeoutEx) {
- if (isLogPHIEnabled(log)) {
- log.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
outOfFrameData.toString().replace('\r', '\n'));
} else {
- log.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ LOG.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
}
resetConnection(socket);
throw new MllpCorruptFrameException("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read", outOfFrameData.toByteArray());
} catch (IOException e) {
- if (isLogPHIEnabled(log)) {
- log.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
outOfFrameData.toString().replace('\r', '\n'));
} else {
- log.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ LOG.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
}
resetConnection(socket);
@@ -182,20 +181,20 @@ public final class MllpUtil {
int readByte = socketInputStream.read();
switch (readByte) {
case END_OF_STREAM:
- if (isLogPHIEnabled(log)) {
- log.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
} else {
- log.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data");
+ LOG.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data");
}
resetConnection(socket);
throw new MllpCorruptFrameException("END_OF_STREAM read while looking for the end of the MLLP frame", payload.size() > 0 ? payload.toByteArray() : null);
case START_OF_BLOCK:
- if (isLogPHIEnabled(log)) {
- log.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
} else {
- log.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data");
+ LOG.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data");
}
resetConnection(socket);
@@ -203,11 +202,11 @@ public final class MllpUtil {
throw new MllpCorruptFrameException("A new MLLP frame was opened before the previous frame was closed", payload.size() > 0 ? payload.toByteArray() : null);
case END_OF_BLOCK:
if (END_OF_DATA != socketInputStream.read()) {
- if (isLogPHIEnabled(log)) {
- log.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data: {}",
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data: {}",
payload.toString().replace('\r', '\n'));
} else {
- log.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data");
+ LOG.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data");
}
resetConnection(socket);
@@ -223,13 +222,13 @@ public final class MllpUtil {
}
} catch (SocketTimeoutException timeoutEx) {
if (0 < payload.size()) {
- if (isLogPHIEnabled(log)) {
- log.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
} else {
- log.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data");
+ LOG.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data");
}
} else {
- log.error("Timeout looking for the end of the MLLP frame - resetting connection");
+ LOG.error("Timeout looking for the end of the MLLP frame - resetting connection");
}
resetConnection(socket);
@@ -237,13 +236,13 @@ public final class MllpUtil {
throw new MllpCorruptFrameException("Timeout looking for the end of the MLLP frame", payload.size() > 0 ? payload.toByteArray() : null, timeoutEx);
} catch (IOException ioEx) {
if (0 < payload.size()) {
- if (isLogPHIEnabled(log)) {
- log.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ if (isLogPHIEnabled(LOG)) {
+ LOG.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
} else {
- log.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data");
+ LOG.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data");
}
} else {
- log.error("Exception encountered looking for the end of the MLLP frame - resetting connection");
+ LOG.error("Exception encountered looking for the end of the MLLP frame - resetting connection");
}
resetConnection(socket);
@@ -266,11 +265,11 @@ public final class MllpUtil {
*/
public static void writeFramedPayload(Socket socket, byte[] payload) throws MllpException {
if (socket.isConnected() && !socket.isClosed()) {
- OutputStream outputStream = null;
+ OutputStream outputStream;
try {
outputStream = new BufferedOutputStream(socket.getOutputStream(), payload.length + 4);
} catch (IOException ioEx) {
- log.error("Error Retrieving OutputStream from Socket - resetting connection");
+ LOG.error("Error Retrieving OutputStream from Socket - resetting connection");
resetConnection(socket);
@@ -285,7 +284,7 @@ public final class MllpUtil {
outputStream.write(END_OF_DATA);
outputStream.flush();
} catch (IOException ioEx) {
- log.error("Error writing MLLP payload - resetting connection");
+ LOG.error("Error writing MLLP payload - resetting connection");
resetConnection(socket);
@@ -301,19 +300,19 @@ public final class MllpUtil {
try {
socket.shutdownInput();
} catch (Exception ex) {
- log.warn("Exception encountered shutting down the input stream on the client socket", ex);
+ LOG.warn("Exception encountered shutting down the input stream on the client socket", ex);
}
try {
socket.shutdownOutput();
} catch (Exception ex) {
- log.warn("Exception encountered shutting down the output stream on the client socket", ex);
+ LOG.warn("Exception encountered shutting down the output stream on the client socket", ex);
}
try {
socket.close();
} catch (Exception ex) {
- log.warn("Exception encountered closing the client socket", ex);
+ LOG.warn("Exception encountered closing the client socket", ex);
}
}
}
@@ -324,13 +323,13 @@ public final class MllpUtil {
try {
socket.setSoLinger(true, 0);
} catch (Exception ex) {
- log.warn("Exception encountered setting SO_LINGER to 0 on the socket to force a reset", ex);
+ LOG.warn("Exception encountered setting SO_LINGER to 0 on the socket to force a reset", ex);
}
try {
socket.close();
} catch (Exception ex) {
- log.warn("Exception encountered closing the client socket", ex);
+ LOG.warn("Exception encountered closing the client socket", ex);
}
}
@@ -358,7 +357,6 @@ public final class MllpUtil {
}
private static boolean isLogPHIEnabled(Logger targetLogger) {
- String logPHIProperty = System.getProperty(MllpComponent.MLLP_LOG_PHI_PROPERTY, "true");
if (targetLogger.isDebugEnabled()) {
if (Boolean.parseBoolean(System.getProperty(MllpComponent.MLLP_LOG_PHI_PROPERTY, "true"))) {
return true;
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
index b95264a..265db76 100644
--- a/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
+++ b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
@@ -35,7 +35,7 @@ import static org.apache.camel.component.mllp.MllpEndpoint.SEGMENT_DELIMITER;
* A Camel Processor for generating HL7 Acknowledgements
*/
public class Hl7AcknowledgementGenerator implements Processor {
- Logger log = LoggerFactory.getLogger(this.getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(Hl7AcknowledgementGenerator.class);
String defaultNack = "MSH|^~\\&|||||||NACK||P|2.2" + SEGMENT_DELIMITER
+ "MSA|AR|" + SEGMENT_DELIMITER
@@ -127,7 +127,7 @@ public class Hl7AcknowledgementGenerator implements Processor {
}
if (-1 == msh92start) {
- log.warn("Didn't find component separator for MSH-9.2 - sending ACK in MSH-9");
+ LOG.warn("Didn't find component separator for MSH-9.2 - sending ACK in MSH-9");
} else {
acknowledgement.write(hl7MessageBytes, msh92start, fieldSeparatorIndexes.get(8) - msh92start); // MSH-9.2
}
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/resources/META-INF/LICENSE.txt b/components/camel-mllp/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-mllp/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/resources/META-INF/NOTICE.txt b/components/camel-mllp/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-mllp/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/e5725284/components/camel-mllp/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/resources/log4j.properties b/components/camel-mllp/src/test/resources/log4j.properties
index 29932b4..19ecb1f 100644
--- a/components/camel-mllp/src/test/resources/log4j.properties
+++ b/components/camel-mllp/src/test/resources/log4j.properties
@@ -1,41 +1,42 @@
-###
- # Licensed to the Apache Software Foundation (ASF) under one or more
- # contributor license agreements. See the NOTICE file distributed with
- # this work for additional information regarding copyright ownership.
- # The ASF licenses this file to You under the Apache License, Version 2.0
- # (the "License"); you may not use this file except in compliance with
- # the License. You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- ##
-
-log4j.rootLogger=INFO, console, file
-
-# CONSOLE appender
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.Target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=[%d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX}] %p | %t | %c %n\t%m%n
-
-# File appender - not normally used in testing
+## ------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ------------------------------------------------------------------------
+
+#
+# The logging properties used for testing.
+#
+log4j.rootLogger=INFO, file
+
+#log4j.logger.org.apache.camel.component.mllp=DEBUG
+#log4j.logger.org.apache.camel.test.mllp.PassthroughProcessor=INFO
+#log4j.logger.org.apache.camel.test.junit=WARN
+#log4j.logger.org.apache.camel.test.tcp=WARN
+
+# CONSOLE appender not used by default
+log4j.appender.out=org.apache.log4j.ConsoleAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+#log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
+log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+# File appender
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.layout=org.apache.log4j.PatternLayout
-log4j.appender.file.layout.ConversionPattern=%d %-5p %c{1} - %m %n
+log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
log4j.appender.file.file=target/camel-mllp-test.log
-# log4j.logger.org.apache.camel.component.hl7=TRACE
-log4j.logger.org.apache.camel.component.mllp=INFO
-log4j.logger.org.apache.camel.test.mllp.PassthroughProcessor=INFO
-log4j.logger.org.apache.camel=WARN
-log4j.logger.org.apache.activemq=WARN
-log4j.logger.org.apache.camel.test.junit=WARN
-log4j.logger.org.apache.camel.test.tcp=WARN
[6/6] camel git commit: Include components in release kit
Posted by da...@apache.org.
Include components in release kit
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0fb50fcf
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0fb50fcf
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0fb50fcf
Branch: refs/heads/master
Commit: 0fb50fcf5c79a2c9bc6b2d93cbe46ed2f3e5600e
Parents: e572528
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Dec 30 10:30:34 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Dec 30 10:30:34 2015 +0100
----------------------------------------------------------------------
apache-camel/pom.xml | 8 ++++
.../src/main/descriptors/common-bin.xml | 1 +
parent/pom.xml | 5 +++
.../camel/itest/karaf/CamelIgniteTest.java | 40 ++++++++++++++++++++
.../apache/camel/itest/karaf/CamelMllpTest.java | 40 ++++++++++++++++++++
5 files changed, 94 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0fb50fcf/apache-camel/pom.xml
----------------------------------------------------------------------
diff --git a/apache-camel/pom.xml b/apache-camel/pom.xml
index 6a54092..2e1913c 100644
--- a/apache-camel/pom.xml
+++ b/apache-camel/pom.xml
@@ -345,6 +345,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-ignite</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-infinispan</artifactId>
</dependency>
<dependency>
@@ -524,6 +528,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-mllp</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-mongodb</artifactId>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/0fb50fcf/apache-camel/src/main/descriptors/common-bin.xml
----------------------------------------------------------------------
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index cd70770..08ad0b0 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -100,6 +100,7 @@
<include>org.apache.camel:camel-ibatis</include>
<include>org.apache.camel:camel-ical</include>
<include>org.apache.camel:camel-infinispan</include>
+ <include>org.apache.camel:camel-ignite</include>
<include>org.apache.camel:camel-irc</include>
<include>org.apache.camel:camel-jackson</include>
<include>org.apache.camel:camel-jacksonxml</include>
http://git-wip-us.apache.org/repos/asf/camel/blob/0fb50fcf/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index 3c4cdbc..20ba291 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1020,6 +1020,11 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-ignite</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-irc</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/0fb50fcf/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelIgniteTest.java
----------------------------------------------------------------------
diff --git a/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelIgniteTest.java b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelIgniteTest.java
new file mode 100644
index 0000000..f697ad2
--- /dev/null
+++ b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelIgniteTest.java
@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.itest.karaf;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+@RunWith(PaxExam.class)
+public class CamelIgniteTest extends AbstractFeatureTest {
+
+ public static final String COMPONENT = extractName(CamelIgniteTest.class);
+
+ @Test
+ public void test() throws Exception {
+ testComponent(COMPONENT);
+ }
+
+ @Configuration
+ public static Option[] configure() {
+ return configure(COMPONENT);
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/0fb50fcf/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelMllpTest.java
----------------------------------------------------------------------
diff --git a/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelMllpTest.java b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelMllpTest.java
new file mode 100644
index 0000000..f467666
--- /dev/null
+++ b/tests/camel-itest-karaf/src/test/java/org/apache/camel/itest/karaf/CamelMllpTest.java
@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.itest.karaf;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+@RunWith(PaxExam.class)
+public class CamelMllpTest extends AbstractFeatureTest {
+
+ public static final String COMPONENT = extractName(CamelMllpTest.class);
+
+ @Test
+ public void test() throws Exception {
+ testComponent(COMPONENT);
+ }
+
+ @Configuration
+ public static Option[] configure() {
+ return configure(COMPONENT);
+ }
+
+}
\ No newline at end of file
[3/6] camel git commit: Added camel-mllp component
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
new file mode 100644
index 0000000..362a30f
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpServerConsumer.java
@@ -0,0 +1,550 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.mllp.impl.MllpUtil;
+import org.apache.camel.converter.IOConverter;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.processor.mllp.Hl7AcknowledgementGenerationException;
+import org.apache.camel.processor.mllp.Hl7AcknowledgementGenerator;
+import org.apache.camel.util.IOHelper;
+
+import static org.apache.camel.component.mllp.MllpConstants.*;
+import static org.apache.camel.component.mllp.MllpEndpoint.SEGMENT_DELIMITER;
+import static org.apache.camel.component.mllp.MllpEndpoint.START_OF_BLOCK;
+
+/**
+ * The MLLP consumer.
+ */
+public class MllpTcpServerConsumer extends DefaultConsumer {
+ ServerSocketThread serverSocketThread;
+
+ List<ClientSocketThread> clientThreads = new LinkedList<>();
+
+ private final MllpEndpoint endpoint;
+
+ public MllpTcpServerConsumer(MllpEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ log.trace("MllpTcpServerConsumer(endpoint, processor)");
+
+
+ this.endpoint = endpoint;
+
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ log.debug("doStart() - creating acceptor thread");
+
+ ServerSocket serverSocket = new ServerSocket();
+ if (null != endpoint.receiveBufferSize) {
+ serverSocket.setReceiveBufferSize(endpoint.receiveBufferSize);
+ }
+
+ serverSocket.setReuseAddress(endpoint.reuseAddress);
+
+ // Accept Timeout
+ serverSocket.setSoTimeout(endpoint.acceptTimeout);
+
+ InetSocketAddress socketAddress = new InetSocketAddress(endpoint.getHostname(), endpoint.getPort());
+ serverSocket.bind(socketAddress, endpoint.backlog);
+
+ serverSocketThread = new ServerSocketThread(serverSocket);
+ serverSocketThread.start();
+
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ log.debug("doStop()");
+
+ switch (serverSocketThread.getState()) {
+ case TERMINATED:
+ // This is what we hope for
+ break;
+ case NEW:
+ case RUNNABLE:
+ case BLOCKED:
+ case WAITING:
+ case TIMED_WAITING:
+ default:
+ serverSocketThread.interrupt();
+ break;
+ }
+
+ serverSocketThread = null;
+
+ super.doStop();
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ log.debug("doSuspend()");
+
+ super.doSuspend();
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ log.debug("doResume()");
+
+ super.doSuspend();
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ log.debug("doShutdown()");
+
+ super.doShutdown();
+ }
+
+
+ /**
+ * Nested Class to handle the ServerSocket.accept requests
+ */
+ class ServerSocketThread extends Thread {
+ ServerSocket serverSocket;
+
+ ServerSocketThread(ServerSocket serverSocket) {
+ this.setName(createThreadName(serverSocket));
+
+ this.serverSocket = serverSocket;
+ }
+
+ /**
+ * Derive a thread name from the class name, the component URI and the connection information.
+ * <p/>
+ * The String will in the format <class name>[endpoint key] - [local socket address]
+ *
+ * @return String for thread name
+ */
+ String createThreadName(ServerSocket serverSocket) {
+ // Get the classname without the package. This is a nested class, so we want the parent class name included
+ String fullClassName = this.getClass().getName();
+ String className = fullClassName.substring(fullClassName.lastIndexOf('.') + 1);
+
+ // Get the URI without options
+ String fullEndpointKey = endpoint.getEndpointKey();
+ String endpointKey;
+ if (fullEndpointKey.contains("?")) {
+ endpointKey = fullEndpointKey.substring(0, fullEndpointKey.indexOf('?'));
+ } else {
+ endpointKey = fullEndpointKey;
+ }
+
+ // Now put it all together
+ return String.format("%s[%s] - %s", className, endpointKey, serverSocket.getLocalSocketAddress());
+ }
+
+ /**
+ * The main ServerSocket.accept() loop
+ * <p/>
+ * NOTE: When a connection is received, the Socket is checked after a brief delay in an attempt to determine
+ * if this is a load-balancer probe. The test is done before the ClientSocketThread is created to avoid creating
+ * a large number of short lived threads, which is what can occur if the load balancer polling interval is very
+ * short.
+ */
+ public void run() {
+ log.debug("Starting acceptor thread");
+
+ while (null != serverSocket && serverSocket.isBound() && !serverSocket.isClosed()) {
+ try {
+ /* ? set this here ? */
+ // serverSocket.setSoTimeout( 10000 );
+ // TODO: Need to check maxConnections and figure out what to do when exceeded
+ Socket socket = serverSocket.accept();
+
+ /* Wait a bit and then check and see if the socket is really there - it could be a load balancer
+ pinging the port
+ */
+ Thread.sleep(100);
+ if (socket.isConnected() && !socket.isClosed()) {
+ log.debug("Socket appears to be there - check for available data");
+ InputStream inputStream;
+ try {
+ inputStream = socket.getInputStream();
+ } catch (IOException ioEx) {
+ // Bad Socket -
+ log.warn("Failed to retrieve the InputStream for socket after the initial connection was accepted");
+ MllpUtil.resetConnection(socket);
+ continue;
+ }
+
+ if (0 < inputStream.available()) {
+ // Something is there - start the client thread
+ ClientSocketThread clientThread = new ClientSocketThread(socket, null);
+ clientThreads.add(clientThread);
+ clientThread.start();
+ continue;
+ }
+
+ // The easy check failed - so trigger a blocking read
+ socket.setSoTimeout(100);
+ try {
+ int tmpByte = inputStream.read();
+ socket.setSoTimeout(endpoint.responseTimeout);
+ if (-1 == tmpByte) {
+ log.debug("Socket.read() returned END_OF_STREAM - resetting connection");
+ MllpUtil.resetConnection(socket);
+ } else {
+ ClientSocketThread clientThread = new ClientSocketThread(socket, tmpByte);
+ clientThreads.add(clientThread);
+ clientThread.start();
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ // No data, but the socket is there
+ log.debug("No Data - but the socket is there. Starting ClientSocketThread");
+ ClientSocketThread clientThread = new ClientSocketThread(socket, null);
+ clientThreads.add(clientThread);
+ clientThread.start();
+ }
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ // No new clients
+ log.trace("SocketTimeoutException waiting for new connections - no new connections");
+
+ for (int i = clientThreads.size() - 1; i >= 0; --i) {
+ ClientSocketThread thread = clientThreads.get(i);
+ if (!thread.isAlive()) {
+ clientThreads.remove(i);
+ }
+ }
+ } catch (InterruptedException interruptEx) {
+ log.info("accept loop interrupted - closing ServerSocket");
+ try {
+ serverSocket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing ServerSocket after InterruptedException", ex);
+ }
+ } catch (Exception ex) {
+ log.error("Exception accepting new connection", ex);
+ }
+ }
+ }
+
+ }
+
+ class ClientSocketThread extends Thread {
+ Socket clientSocket;
+ Hl7AcknowledgementGenerator acknowledgementGenerator = new Hl7AcknowledgementGenerator();
+
+ Integer initialByte;
+
+ ClientSocketThread(Socket clientSocket, Integer initialByte) throws IOException {
+ this.initialByte = initialByte;
+ this.setName(createThreadName(clientSocket));
+ this.clientSocket = clientSocket;
+ this.clientSocket.setKeepAlive(endpoint.keepAlive);
+ this.clientSocket.setTcpNoDelay(endpoint.tcpNoDelay);
+ if (null != endpoint.receiveBufferSize) {
+ this.clientSocket.setReceiveBufferSize(endpoint.receiveBufferSize);
+ }
+ if (null != endpoint.sendBufferSize) {
+ this.clientSocket.setSendBufferSize(endpoint.sendBufferSize);
+ }
+ this.clientSocket.setReuseAddress(endpoint.reuseAddress);
+ this.clientSocket.setSoLinger(false, -1);
+
+ // Read Timeout
+ this.clientSocket.setSoTimeout(endpoint.responseTimeout);
+
+ }
+
+ /**
+ * derive a thread name from the class name, the component URI and the connection information
+ * <p/>
+ * The String will in the format <class name>[endpoint key] - [local socket address] -> [remote socket address]
+ *
+ * @return the thread name
+ */
+ String createThreadName(Socket socket) {
+ // Get the classname without the package. This is a nested class, so we want the parent class name included
+ String fullClassName = this.getClass().getName();
+ String className = fullClassName.substring(fullClassName.lastIndexOf('.') + 1);
+
+ // Get the URI without options
+ String fullEndpointKey = endpoint.getEndpointKey();
+ String endpointKey;
+ if (fullEndpointKey.contains("?")) {
+ endpointKey = fullEndpointKey.substring(0, fullEndpointKey.indexOf('?'));
+ } else {
+ endpointKey = fullEndpointKey;
+ }
+
+ // Now put it all together
+ return String.format("%s[%s] - %s -> %s", className, endpointKey, socket.getLocalSocketAddress(), socket.getRemoteSocketAddress());
+ }
+
+ @Override
+ public void run() {
+
+ while (null != clientSocket && clientSocket.isConnected() && !clientSocket.isClosed()) {
+ byte[] hl7MessageBytes = null;
+ // Send the message on for processing and wait for the response
+ log.debug("Reading data ....");
+ try {
+ if (null != initialByte && START_OF_BLOCK == initialByte) {
+ hl7MessageBytes = MllpUtil.closeFrame(clientSocket);
+ } else {
+ try {
+ MllpUtil.openFrame(clientSocket);
+ } catch (SocketTimeoutException timeoutEx) {
+ // When thrown by openFrame, it indicates that no data was available - but no error
+ continue;
+ }
+ hl7MessageBytes = MllpUtil.closeFrame(clientSocket);
+ }
+ } catch (MllpException mllpEx) {
+ Exchange exchange = endpoint.createExchange(ExchangePattern.InOut);
+ exchange.setException(mllpEx);
+ return;
+ } finally {
+ initialByte = null;
+ }
+
+ if (null == hl7MessageBytes) {
+ continue;
+ }
+
+ log.debug("Populating the exchange with received message");
+ Exchange exchange = endpoint.createExchange(ExchangePattern.InOut);
+ Message message = exchange.getIn();
+ message.setBody(hl7MessageBytes, byte[].class);
+
+ message.setHeader(MLLP_LOCAL_ADDRESS, clientSocket.getLocalAddress().toString());
+ message.setHeader(MLLP_REMOTE_ADDRESS, clientSocket.getRemoteSocketAddress());
+
+ populateHl7DataHeaders(exchange, message, hl7MessageBytes);
+
+
+ log.debug("Calling processor");
+ try {
+ getProcessor().process(exchange);
+ // processed the message - send the acknowledgement
+
+ // Check BEFORE_SEND Properties
+ if (exchange.getProperty(MLLP_RESET_CONNECTION_BEFORE_SEND, boolean.class)) {
+ MllpUtil.resetConnection(clientSocket);
+ return;
+ } else if (exchange.getProperty(MLLP_CLOSE_CONNECTION_BEFORE_SEND, boolean.class)) {
+ MllpUtil.closeConnection(clientSocket);
+ }
+
+ // Find the acknowledgement body
+ byte[] acknowledgementMessageBytes = exchange.getProperty(MLLP_ACKNOWLEDGEMENT, byte[].class);
+ String acknowledgementMessageType = null;
+ if (null == acknowledgementMessageBytes) {
+ if (!endpoint.autoAck) {
+ exchange.setException(new MllpInvalidAcknowledgementException("Automatic Acknowledgement is disabled and the "
+ + MLLP_ACKNOWLEDGEMENT + " exchange property is null or cannot be converted to byte[]"));
+ return;
+ }
+
+ String acknowledgmentTypeProperty = exchange.getProperty(MLLP_ACKNOWLEDGEMENT_TYPE, String.class);
+ try {
+ if (null == acknowledgmentTypeProperty) {
+ if (null == exchange.getException()) {
+ acknowledgementMessageType = "AA";
+ acknowledgementMessageBytes = acknowledgementGenerator.generateApplicationAcceptAcknowledgementMessage(hl7MessageBytes);
+ } else {
+ acknowledgementMessageType = "AE";
+ acknowledgementMessageBytes = acknowledgementGenerator.generateApplicationErrorAcknowledgementMessage(hl7MessageBytes);
+ }
+ } else {
+ switch (acknowledgmentTypeProperty) {
+ case "AA":
+ acknowledgementMessageType = "AA";
+ acknowledgementMessageBytes = acknowledgementGenerator.generateApplicationAcceptAcknowledgementMessage(hl7MessageBytes);
+ break;
+ case "AE":
+ acknowledgementMessageType = "AE";
+ acknowledgementMessageBytes = acknowledgementGenerator.generateApplicationErrorAcknowledgementMessage(hl7MessageBytes);
+ break;
+ case "AR":
+ acknowledgementMessageType = "AR";
+ acknowledgementMessageBytes = acknowledgementGenerator.generateApplicationRejectAcknowledgementMessage(hl7MessageBytes);
+ break;
+ default:
+ exchange.setException(new Hl7AcknowledgementGenerationException("Unsupported acknowledgment type: " + acknowledgmentTypeProperty));
+ return;
+ }
+ }
+ } catch (Hl7AcknowledgementGenerationException ackGenerationException) {
+ exchange.setException(ackGenerationException);
+ }
+ } else {
+ final byte bM = 77;
+ final byte bS = 83;
+ final byte bA = 65;
+ final byte bE = 69;
+ final byte bR = 82;
+
+ final byte fieldSeparator = hl7MessageBytes[3];
+ // Acknowledgment is specified in exchange property - determine the acknowledgement type
+ for (int i = 0; i < hl7MessageBytes.length; ++i) {
+ if (SEGMENT_DELIMITER == i) {
+ if (i + 7 < hl7MessageBytes.length // Make sure we don't run off the end of the message
+ && bM == hl7MessageBytes[i + 1] && bS == hl7MessageBytes[i + 2] && bA == hl7MessageBytes[i + 3] && fieldSeparator == hl7MessageBytes[i + 4]) {
+ if (fieldSeparator != hl7MessageBytes[i + 7]) {
+ log.warn("MSA-1 is longer than 2-bytes - ignoring trailing bytes");
+ }
+ // Found MSA - pull acknowledgement bytes
+ byte[] acknowledgmentTypeBytes = new byte[2];
+ acknowledgmentTypeBytes[0] = hl7MessageBytes[i + 5];
+ acknowledgmentTypeBytes[1] = hl7MessageBytes[i + 6];
+ acknowledgementMessageType = IOConverter.toString(acknowledgmentTypeBytes, exchange);
+
+ // Verify it's a valid acknowledgement code
+ if (bA != acknowledgmentTypeBytes[0]) {
+ switch (acknowledgementMessageBytes[1]) {
+ case bA:
+ case bR:
+ case bE:
+ break;
+ default:
+ log.warn("Invalid acknowledgement type [" + acknowledgementMessageType + "] found in message - should be AA, AE or AR");
+ }
+ }
+
+ // if the MLLP_ACKNOWLEDGEMENT_TYPE property is set on the exchange, make sure it matches
+ String acknowledgementTypeProperty = exchange.getProperty(MLLP_ACKNOWLEDGEMENT_TYPE, String.class);
+ if (null != acknowledgementTypeProperty && !acknowledgementTypeProperty.equals(acknowledgementMessageType)) {
+ log.warn("Acknowledgement type found in message [" + acknowledgementMessageType + "] does not match "
+ + MLLP_ACKNOWLEDGEMENT_TYPE + " exchange property value [" + acknowledgementTypeProperty + "] - using value found in message");
+ }
+ }
+ }
+ }
+ }
+
+ // Send the acknowledgement
+ log.debug("Writing Acknowledgement");
+ MllpUtil.writeFramedPayload(clientSocket, acknowledgementMessageBytes);
+ exchange.getIn().setHeader(MLLP_ACKNOWLEDGEMENT, acknowledgementMessageBytes);
+ exchange.getIn().setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, acknowledgementMessageType);
+
+ // Check AFTER_SEND Properties
+ if (exchange.getProperty(MLLP_RESET_CONNECTION_AFTER_SEND, boolean.class)) {
+ MllpUtil.resetConnection(clientSocket);
+ return;
+ } else if (exchange.getProperty(MLLP_CLOSE_CONNECTION_AFTER_SEND, boolean.class)) {
+ MllpUtil.closeConnection(clientSocket);
+ }
+
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+
+ }
+
+ log.info("ClientSocketThread exiting");
+
+ }
+
+ private void populateHl7DataHeaders(Exchange exchange, Message message, byte[] hl7MessageBytes) {
+ // Find the end of the MSH and indexes of the fields in the MSH to populate message headers
+ final byte fieldSeparator = hl7MessageBytes[3];
+ final byte componentSeparator = hl7MessageBytes[4];
+ int endOfMSH = -1;
+ List<Integer> fieldSeparatorIndexes = new ArrayList<>(10); // We need at least 10 fields to create the acknowledgment
+
+ for (int i = 0; i < hl7MessageBytes.length; ++i) {
+ if (fieldSeparator == hl7MessageBytes[i]) {
+ fieldSeparatorIndexes.add(i);
+ } else if (SEGMENT_DELIMITER == hl7MessageBytes[i]) {
+ endOfMSH = i;
+ break;
+ }
+ }
+
+ if (-1 == endOfMSH) {
+ // TODO: May want to throw some sort of an Exception here
+ log.error("Population of message headers failed - unable to find the end of the MSH segment");
+ } else {
+ log.debug("Populating the message headers");
+ Charset charset = Charset.forName(IOHelper.getCharsetName(exchange));
+
+ // MSH-3
+ message.setHeader(MLLP_SENDING_APPLICATION, new String(hl7MessageBytes, fieldSeparatorIndexes.get(1) + 1,
+ fieldSeparatorIndexes.get(2) - fieldSeparatorIndexes.get(1) - 1, charset));
+ // MSH-4
+ message.setHeader(MLLP_SENDING_FACILITY, new String(hl7MessageBytes, fieldSeparatorIndexes.get(2) + 1,
+ fieldSeparatorIndexes.get(3) - fieldSeparatorIndexes.get(2) - 1, charset));
+ // MSH-5
+ message.setHeader(MLLP_RECEIVING_APPLICATION, new String(hl7MessageBytes, fieldSeparatorIndexes.get(3) + 1,
+ fieldSeparatorIndexes.get(4) - fieldSeparatorIndexes.get(3) - 1,
+ charset));
+ // MSH-6
+ message.setHeader(MLLP_RECEIVING_FACILITY, new String(hl7MessageBytes, fieldSeparatorIndexes.get(4) + 1,
+ fieldSeparatorIndexes.get(5) - fieldSeparatorIndexes.get(4) - 1,
+ charset));
+ // MSH-7
+ message.setHeader(MLLP_TIMESTAMP, new String(hl7MessageBytes, fieldSeparatorIndexes.get(5) + 1,
+ fieldSeparatorIndexes.get(6) - fieldSeparatorIndexes.get(5) - 1, charset));
+ // MSH-8
+ message.setHeader(MLLP_SECURITY, new String(hl7MessageBytes, fieldSeparatorIndexes.get(6) + 1,
+ fieldSeparatorIndexes.get(7) - fieldSeparatorIndexes.get(6) - 1, charset));
+ // MSH-9
+ message.setHeader(MLLP_MESSAGE_TYPE, new String(hl7MessageBytes, fieldSeparatorIndexes.get(7) + 1,
+ fieldSeparatorIndexes.get(8) - fieldSeparatorIndexes.get(7) - 1, charset));
+ // MSH-10
+ message.setHeader(MLLP_MESSAGE_CONTROL, new String(hl7MessageBytes, fieldSeparatorIndexes.get(8) + 1,
+ fieldSeparatorIndexes.get(9) - fieldSeparatorIndexes.get(8) - 1, charset));
+ // MSH-11
+ message.setHeader(MLLP_PROCESSING_ID, new String(hl7MessageBytes, fieldSeparatorIndexes.get(9) + 1,
+ fieldSeparatorIndexes.get(10) - fieldSeparatorIndexes.get(9) - 1, charset));
+ // MSH-12
+ message.setHeader(MLLP_VERSION_ID, new String(hl7MessageBytes, fieldSeparatorIndexes.get(10) + 1,
+ fieldSeparatorIndexes.get(11) - fieldSeparatorIndexes.get(10) - 1, charset));
+ // MSH-18
+ message.setHeader(MLLP_CHARSET, new String(hl7MessageBytes, fieldSeparatorIndexes.get(16) + 1,
+ fieldSeparatorIndexes.get(17) - fieldSeparatorIndexes.get(16) - 1, charset));
+
+ for (int i = fieldSeparatorIndexes.get(7) + 1; i < fieldSeparatorIndexes.get(8); ++i) {
+ if (componentSeparator == hl7MessageBytes[i]) {
+ // MSH-9.1
+ message.setHeader(MLLP_EVENT_TYPE, new String(hl7MessageBytes, fieldSeparatorIndexes.get(7) + 1,
+ i - fieldSeparatorIndexes.get(7) - 1, charset));
+ // MSH-9.2
+ message.setHeader(MLLP_TRIGGER_EVENT, new String(hl7MessageBytes, i + 1,
+ fieldSeparatorIndexes.get(8) - i - 1, charset));
+ break;
+ }
+ }
+ }
+ }
+
+
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTimeoutException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTimeoutException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTimeoutException.java
new file mode 100644
index 0000000..0e54772
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTimeoutException.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer or Consumer encounter a timeout reading a message
+ */
+public class MllpTimeoutException extends MllpException {
+ public MllpTimeoutException(String message) {
+ super(message);
+ }
+
+ public MllpTimeoutException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpTimeoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpTimeoutException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpWriteException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpWriteException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpWriteException.java
new file mode 100644
index 0000000..43fe740
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpWriteException.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer or consumer encounter an error transmitting data
+ */
+public class MllpWriteException extends MllpException {
+ public MllpWriteException(String message) {
+ super(message);
+ }
+
+ public MllpWriteException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpWriteException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpWriteException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
new file mode 100644
index 0000000..4fd3fbf
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/impl/MllpUtil.java
@@ -0,0 +1,371 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp.impl;
+
+import java.io.BufferedOutputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+
+import org.apache.camel.component.mllp.MllpComponent;
+import org.apache.camel.component.mllp.MllpCorruptFrameException;
+import org.apache.camel.component.mllp.MllpException;
+import org.apache.camel.component.mllp.MllpTimeoutException;
+import org.apache.camel.component.mllp.MllpWriteException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_BLOCK;
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_DATA;
+import static org.apache.camel.component.mllp.MllpEndpoint.END_OF_STREAM;
+import static org.apache.camel.component.mllp.MllpEndpoint.START_OF_BLOCK;
+
+
+/**
+ * Supplies methods to read and write messages in a MLLP Frame.
+ * <p/>
+ * Although the methods in the class are intended to handle HL7 v2 formatted messages, the methods do not
+ * depend on that format - any byte[]can be written to the Socket. Also, any byte[] can be read from the socket
+ * provided it has the proper MLLP Enveloping - <START_OF_BLOCK>payload<END_OF_BLOCK><END_OF_DATA>>.
+ * <p/>
+ * NOTE: MLLP payloads are not logged unless the logging level is set to DEBUG or TRACE to avoid introducing PHI
+ * into the log files. Logging of PHI can be globally disabled by setting the org.apache.camel.mllp.logPHI system
+ * property. The property is evaluated using Boolean.parseBoolean.
+ * <p/>
+ */
+public final class MllpUtil {
+ static Logger log = LoggerFactory.getLogger(MllpUtil.class);
+
+ private MllpUtil() {
+ }
+
+ /**
+ * Open the MLLP frame by reading from the Socket until the begging of the frame is found.
+ * <p/>
+ * If any errors occur (including MLLP frame errors) while opening the frame, the socket will be closed and an
+ * Exception will be thrown.
+ *
+ * @param socket the Socket to read
+ * @throws SocketTimeoutException thrown if a timeout occurs while looking for the beginning of the MLLP frame, but
+ * nothing is yet available - this is NOT an error condition
+ * @throws MllpCorruptFrameException if the MLLP Frame is corrupted in some way
+ * @throws MllpException for other unexpected error conditions
+ */
+ public static void openFrame(Socket socket) throws SocketTimeoutException, MllpCorruptFrameException, MllpException {
+ if (socket.isConnected() && !socket.isClosed()) {
+ InputStream socketInputStream = MllpUtil.getInputStream(socket);
+
+ int readByte;
+ try {
+ readByte = socketInputStream.read();
+ switch (readByte) {
+ case START_OF_BLOCK:
+ return;
+ case END_OF_STREAM:
+ resetConnection(socket);
+ return;
+ default:
+ // Continue on and process the out-of-frame data
+ }
+ } catch (SocketTimeoutException normaTimeoutEx) {
+ // Just pass this on - the caller will wrap it in a MllpTimeoutException
+ throw normaTimeoutEx;
+ } catch (IOException unexpectedException) {
+ log.error("Unexpected Exception occurred opening MLLP frame - resetting the connection");
+ MllpUtil.resetConnection(socket);
+ throw new MllpException("Unexpected Exception occurred opening MLLP frame", unexpectedException);
+ }
+
+ /*
+ From here on, we're in a bad frame state. Read what's left in the socket, close the connection and
+ return the out-of-frame data.
+ */
+ ByteArrayOutputStream outOfFrameData = new ByteArrayOutputStream();
+ outOfFrameData.write(readByte);
+
+ try {
+ while (true) {
+ readByte = socketInputStream.read();
+ switch (readByte) {
+ case END_OF_STREAM:
+ if (isLogPHIEnabled(log)) {
+ log.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and "
+ + "out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
+ outOfFrameData.toString().replace('\r', '\n'));
+ } else {
+ log.error("END_OF_STREAM read while looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ }
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("END_OF_STREAM read while looking for the beginning of the MLLP frame", outOfFrameData.toByteArray());
+ case START_OF_BLOCK:
+ if (isLogPHIEnabled(log)) {
+ log.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data: {}", outOfFrameData.toString().replace('\r', '\n'));
+ } else {
+ log.warn("The beginning of the MLLP frame was preceded by out-of-frame data - eating data");
+ }
+
+ throw new MllpCorruptFrameException("The beginning of the MLLP frame was preceded by out-of-frame data", outOfFrameData.toByteArray());
+ default:
+ // still reading out-of-frame data
+ outOfFrameData.write(readByte);
+ break;
+ }
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ if (isLogPHIEnabled(log)) {
+ log.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
+ outOfFrameData.toString().replace('\r', '\n'));
+ } else {
+ log.error("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("Timeout looking for the beginning of the MLLP frame, and out-of-frame data had been read", outOfFrameData.toByteArray());
+ } catch (IOException e) {
+ if (isLogPHIEnabled(log)) {
+ log.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data: {}",
+ outOfFrameData.toString().replace('\r', '\n'));
+ } else {
+ log.error("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read - resetting connection and eating out-of-frame data");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("Exception encountered looking for the beginning of the MLLP frame, and out-of-frame data had been read", outOfFrameData.toByteArray());
+ }
+ }
+ }
+
+ /**
+ * Close a MLLP frame by reading from the socket until the end of the frame is found.
+ * <p/>
+ * The method assumes the MLLP frame has already been opened and the first byte available
+ * will be the first byte of the framed message.
+ * <p/>
+ * The method consumes the END_OF_BLOCK and END_OF_DATA bytes from the stream before returning the payload
+ * <p/>
+ * If any errors occur (including MLLP frame errors) while opening the frame, the socket will be closed and an
+ * Exception will be thrown.
+ *
+ * @param socket the Socket to be read
+ * @return the payload of the MLLP-Enveloped message as a byte[]
+ * @throws MllpTimeoutException thrown if a timeout occurs while closing the MLLP frame
+ * @throws MllpCorruptFrameException if the MLLP Frame is corrupted in some way
+ * @throws MllpException for other unexpected error conditions
+ */
+ public static byte[] closeFrame(Socket socket) throws MllpTimeoutException, MllpCorruptFrameException, MllpException {
+ if (socket.isConnected() && !socket.isClosed()) {
+ InputStream socketInputStream = MllpUtil.getInputStream(socket);
+ // TODO: Come up with an intelligent way to size this stream
+ ByteArrayOutputStream payload = new ByteArrayOutputStream(4096);
+ try {
+ while (true) {
+ int readByte = socketInputStream.read();
+ switch (readByte) {
+ case END_OF_STREAM:
+ if (isLogPHIEnabled(log)) {
+ log.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ } else {
+ log.error("END_OF_STREAM read while looking for the end of the MLLP frame - resetting connection and eating data");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("END_OF_STREAM read while looking for the end of the MLLP frame", payload.size() > 0 ? payload.toByteArray() : null);
+ case START_OF_BLOCK:
+ if (isLogPHIEnabled(log)) {
+ log.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ } else {
+ log.error("A new MLLP frame was opened before the previous frame was closed - resetting connection and eating data");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("A new MLLP frame was opened before the previous frame was closed", payload.size() > 0 ? payload.toByteArray() : null);
+ case END_OF_BLOCK:
+ if (END_OF_DATA != socketInputStream.read()) {
+ if (isLogPHIEnabled(log)) {
+ log.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data: {}",
+ payload.toString().replace('\r', '\n'));
+ } else {
+ log.error("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA - resetting connection and eating data");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("The MLLP frame was partially closed - END_OF_BLOCK was not followed by END_OF_DATA",
+ payload.size() > 0 ? payload.toByteArray() : null);
+ }
+ return payload.toByteArray();
+ default:
+ // log.trace( "Read Character: {}", (char)readByte );
+ payload.write(readByte);
+ }
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ if (0 < payload.size()) {
+ if (isLogPHIEnabled(log)) {
+ log.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ } else {
+ log.error("Timeout looking for the end of the MLLP frame - resetting connection and eating data");
+ }
+ } else {
+ log.error("Timeout looking for the end of the MLLP frame - resetting connection");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpCorruptFrameException("Timeout looking for the end of the MLLP frame", payload.size() > 0 ? payload.toByteArray() : null, timeoutEx);
+ } catch (IOException ioEx) {
+ if (0 < payload.size()) {
+ if (isLogPHIEnabled(log)) {
+ log.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data: {}", payload.toString().replace('\r', '\n'));
+ } else {
+ log.error("Exception encountered looking for the end of the MLLP frame - resetting connection and eating data");
+ }
+ } else {
+ log.error("Exception encountered looking for the end of the MLLP frame - resetting connection");
+ }
+
+ resetConnection(socket);
+
+ throw new MllpException("Exception encountered looking for the end of the MLLP frame", payload.size() > 0 ? payload.toByteArray() : null, ioEx);
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * Write a MLLP-Framed payload to the Socket
+ *
+ * @param socket the Socket to write the payload
+ * @param payload the MLLP payload
+ * @return true if write was successful; false otherwise
+ * @throws MllpWriteException if the write fails
+ * @throws MllpException for other unexpected error conditions
+ */
+ public static void writeFramedPayload(Socket socket, byte[] payload) throws MllpException {
+ if (socket.isConnected() && !socket.isClosed()) {
+ OutputStream outputStream = null;
+ try {
+ outputStream = new BufferedOutputStream(socket.getOutputStream(), payload.length + 4);
+ } catch (IOException ioEx) {
+ log.error("Error Retrieving OutputStream from Socket - resetting connection");
+
+ resetConnection(socket);
+
+ throw new MllpException("Error Retrieving OutputStream from Socket", ioEx);
+ }
+
+ if (null != outputStream) {
+ try {
+ outputStream.write(START_OF_BLOCK);
+ outputStream.write(payload, 0, payload.length);
+ outputStream.write(END_OF_BLOCK);
+ outputStream.write(END_OF_DATA);
+ outputStream.flush();
+ } catch (IOException ioEx) {
+ log.error("Error writing MLLP payload - resetting connection");
+
+ resetConnection(socket);
+
+ throw new MllpWriteException("Error writing MLLP payload", payload, ioEx);
+ }
+ }
+ }
+ }
+
+ public static void closeConnection(Socket socket) {
+ if (null != socket) {
+ if (!socket.isClosed()) {
+ try {
+ socket.shutdownInput();
+ } catch (Exception ex) {
+ log.warn("Exception encountered shutting down the input stream on the client socket", ex);
+ }
+
+ try {
+ socket.shutdownOutput();
+ } catch (Exception ex) {
+ log.warn("Exception encountered shutting down the output stream on the client socket", ex);
+ }
+
+ try {
+ socket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing the client socket", ex);
+ }
+ }
+ }
+ }
+
+ public static void resetConnection(Socket socket) {
+ if (null != socket) {
+ try {
+ socket.setSoLinger(true, 0);
+ } catch (Exception ex) {
+ log.warn("Exception encountered setting SO_LINGER to 0 on the socket to force a reset", ex);
+ }
+
+ try {
+ socket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing the client socket", ex);
+ }
+
+ }
+
+ }
+
+ /**
+ * Retrieve the InputStream from the Socket
+ * <p/>
+ * Private utility method that catches IOExceptions when retrieving the InputStream
+ *
+ * @param socket Socket to get the InputStream from
+ * @return the InputStream for the Socket
+ * @throws MllpException when unexpected conditions occur
+ */
+ private static InputStream getInputStream(Socket socket) throws MllpException {
+ InputStream socketInputStream = null;
+ try {
+ socketInputStream = socket.getInputStream();
+ } catch (IOException ioEx) {
+ throw new MllpException("Error Retrieving InputStream from Socket", ioEx);
+ }
+
+ return socketInputStream;
+ }
+
+ private static boolean isLogPHIEnabled(Logger targetLogger) {
+ String logPHIProperty = System.getProperty(MllpComponent.MLLP_LOG_PHI_PROPERTY, "true");
+ if (targetLogger.isDebugEnabled()) {
+ if (Boolean.parseBoolean(System.getProperty(MllpComponent.MLLP_LOG_PHI_PROPERTY, "true"))) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerationException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerationException.java b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerationException.java
new file mode 100644
index 0000000..c27c7f9
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerationException.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.mllp;
+
+public class Hl7AcknowledgementGenerationException extends Exception {
+ private final byte[] hl7Message;
+
+ public Hl7AcknowledgementGenerationException(String message) {
+ super(message);
+ this.hl7Message = null;
+ }
+
+ public Hl7AcknowledgementGenerationException(String message, byte[] hl7Message) {
+ super(message);
+ this.hl7Message = hl7Message;
+ }
+
+ public Hl7AcknowledgementGenerationException(String message, byte[] hl7Message, Throwable cause) {
+ super(message, cause);
+ this.hl7Message = hl7Message;
+ }
+
+
+ public byte[] getHl7Message() {
+ return hl7Message;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
new file mode 100644
index 0000000..b95264a
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/processor/mllp/Hl7AcknowledgementGenerator.java
@@ -0,0 +1,150 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.processor.mllp;
+
+import java.io.ByteArrayOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_ACKNOWLEDGEMENT;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_ACKNOWLEDGEMENT_TYPE;
+import static org.apache.camel.component.mllp.MllpEndpoint.MESSAGE_TERMINATOR;
+import static org.apache.camel.component.mllp.MllpEndpoint.SEGMENT_DELIMITER;
+
+/**
+ * A Camel Processor for generating HL7 Acknowledgements
+ */
+public class Hl7AcknowledgementGenerator implements Processor {
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ String defaultNack = "MSH|^~\\&|||||||NACK||P|2.2" + SEGMENT_DELIMITER
+ + "MSA|AR|" + SEGMENT_DELIMITER
+ + MESSAGE_TERMINATOR;
+
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ Message message = null;
+ if (exchange.hasOut()) {
+ message = exchange.getOut();
+ } else {
+ message = exchange.getIn();
+ }
+
+ byte[] hl7Bytes = message.getMandatoryBody(byte[].class);
+
+ byte[] acknowledgementBytes = null;
+ if (null == exchange.getException()) {
+ acknowledgementBytes = generateApplicationAcceptAcknowledgementMessage(hl7Bytes);
+ message.setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, "AA");
+ } else {
+ acknowledgementBytes = generateApplicationErrorAcknowledgementMessage(hl7Bytes);
+ message.setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, "AE");
+ }
+
+ message.setHeader(MLLP_ACKNOWLEDGEMENT, acknowledgementBytes);
+ }
+
+ public byte[] generateApplicationAcceptAcknowledgementMessage(byte[] hl7MessageBytes) throws Hl7AcknowledgementGenerationException {
+ return generateAcknowledgementMessage(hl7MessageBytes, "AA");
+ }
+
+ public byte[] generateApplicationRejectAcknowledgementMessage(byte[] hl7MessageBytes) throws Hl7AcknowledgementGenerationException {
+ return generateAcknowledgementMessage(hl7MessageBytes, "AR");
+ }
+
+ public byte[] generateApplicationErrorAcknowledgementMessage(byte[] hl7MessageBytes) throws Hl7AcknowledgementGenerationException {
+ return generateAcknowledgementMessage(hl7MessageBytes, "AE");
+ }
+
+ byte[] generateAcknowledgementMessage(byte[] hl7MessageBytes, String acknowledgementCode) throws Hl7AcknowledgementGenerationException {
+ if (hl7MessageBytes == null) {
+ throw new Hl7AcknowledgementGenerationException("Null HL7 message received for parsing operation");
+ }
+
+ final byte fieldSeparator = hl7MessageBytes[3];
+ final byte componentSeparator = hl7MessageBytes[4];
+
+ List<Integer> fieldSeparatorIndexes = new ArrayList<>(10); // We need at least 10 fields to create the acknowledgment
+
+ // Find the end of the MSH and indexes of the fields in the MSH
+ int endOfMSH = -1;
+ for (int i = 0; i < hl7MessageBytes.length; ++i) {
+ if (fieldSeparator == hl7MessageBytes[i]) {
+ fieldSeparatorIndexes.add(i);
+ } else if (SEGMENT_DELIMITER == hl7MessageBytes[i]) {
+ endOfMSH = i;
+ break;
+ }
+ }
+
+ if (-1 == endOfMSH) {
+ throw new Hl7AcknowledgementGenerationException("Failed to find the end of the MSH Segment while attempting to generate response", hl7MessageBytes);
+ }
+
+ if (8 > fieldSeparatorIndexes.size()) {
+ throw new Hl7AcknowledgementGenerationException("Insufficient number of fields in after MSH-2 in MSH to generate a response - 8 are required but "
+ + fieldSeparatorIndexes.size() + " " + "were found", hl7MessageBytes);
+ }
+
+ // Build the MSH Segment
+ ByteArrayOutputStream acknowledgement = new ByteArrayOutputStream(1024);
+ acknowledgement.write(hl7MessageBytes, 0, 8); // through MSH-2 (without trailing field separator)
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(3), fieldSeparatorIndexes.get(4) - fieldSeparatorIndexes.get(3)); // MSH-5
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(4), fieldSeparatorIndexes.get(5) - fieldSeparatorIndexes.get(4)); // MSH-6
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(1), fieldSeparatorIndexes.get(2) - fieldSeparatorIndexes.get(1)); // MSH-3
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(2), fieldSeparatorIndexes.get(3) - fieldSeparatorIndexes.get(2)); // MSH-4
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(5), fieldSeparatorIndexes.get(7) - fieldSeparatorIndexes.get(5)); // MSH-7 and MSH-8
+ // Need to generate the correct MSH-9
+ acknowledgement.write(fieldSeparator);
+ acknowledgement.write("ACK".getBytes(), 0, 3); // MSH-9.1
+ int msh92start = -1;
+ for (int j = fieldSeparatorIndexes.get(7) + 1; j < fieldSeparatorIndexes.get(8); ++j) {
+ if (componentSeparator == hl7MessageBytes[j]) {
+ msh92start = j;
+ break;
+ }
+ }
+
+ if (-1 == msh92start) {
+ log.warn("Didn't find component separator for MSH-9.2 - sending ACK in MSH-9");
+ } else {
+ acknowledgement.write(hl7MessageBytes, msh92start, fieldSeparatorIndexes.get(8) - msh92start); // MSH-9.2
+ }
+
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(8), endOfMSH - fieldSeparatorIndexes.get(8)); // MSH-10 through the end of the MSH
+ acknowledgement.write(SEGMENT_DELIMITER);
+
+ // Build the MSA Segment
+ acknowledgement.write("MSA".getBytes(), 0, 3);
+ acknowledgement.write(fieldSeparator);
+ acknowledgement.write(acknowledgementCode.getBytes(), 0, 2);
+ acknowledgement.write(hl7MessageBytes, fieldSeparatorIndexes.get(8), fieldSeparatorIndexes.get(9) - fieldSeparatorIndexes.get(8)); // MSH-10 end
+ acknowledgement.write(SEGMENT_DELIMITER);
+
+ // Terminate the message
+ acknowledgement.write(MESSAGE_TERMINATOR);
+
+ return acknowledgement.toByteArray();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/resources/META-INF/services/org/apache/camel/component/mllp
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/resources/META-INF/services/org/apache/camel/component/mllp b/components/camel-mllp/src/main/resources/META-INF/services/org/apache/camel/component/mllp
new file mode 100644
index 0000000..3aedcac
--- /dev/null
+++ b/components/camel-mllp/src/main/resources/META-INF/services/org/apache/camel/component/mllp
@@ -0,0 +1 @@
+class=org.apache.camel.component.mllp.MllpComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpProducerConsumerLoopbackTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpProducerConsumerLoopbackTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpProducerConsumerLoopbackTest.java
new file mode 100644
index 0000000..027de7c
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpProducerConsumerLoopbackTest.java
@@ -0,0 +1,121 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.test.mllp.PassthroughProcessor;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+
+public class MllpProducerConsumerLoopbackTest extends CamelTestSupport {
+ int mllpPort = AvailablePortFinder.getNextAvailable();
+
+ @EndpointInject(uri = "direct://source")
+ ProducerTemplate source;
+
+ @EndpointInject(uri = "mock://acknowledged")
+ MockEndpoint acknowledged;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext) super.createCamelContext();
+
+ context.setUseMDCLogging(true);
+ context.setName(this.getClass().getSimpleName());
+
+ return context;
+ }
+
+
+ @Override
+ protected RouteBuilder[] createRouteBuilders() throws Exception {
+ RouteBuilder[] builders = new RouteBuilder[2];
+ final int groupInterval = 1000;
+ final boolean groupActiveOnly = false;
+
+ builders[0] = new RouteBuilder() {
+ String routeId = "mllp-receiver";
+
+ public void configure() {
+ fromF("mllp:%d?autoAck=true", mllpPort)
+ .convertBodyTo(String.class)
+ .to(acknowledged)
+ .process(new PassthroughProcessor("after send to result"))
+ .log(LoggingLevel.DEBUG, routeId, "Receiving: ${body}")
+ .toF("log://%s?level=INFO&groupInterval=%d&groupActiveOnly=%b", routeId, groupInterval, groupActiveOnly);
+ }
+ };
+
+ builders[1] = new RouteBuilder() {
+ String routeId = "mllp-sender";
+
+ String host = "0.0.0.0";
+
+ public void configure() {
+ from(source.getDefaultEndpoint()).routeId(routeId)
+ .log(LoggingLevel.DEBUG, routeId, "Sending: ${body}")
+ .toF("mllp://%s:%d", host, mllpPort)
+ .setBody(header(MllpConstants.MLLP_ACKNOWLEDGEMENT))
+ .toF("log://%s?level=INFO&groupInterval=%d&groupActiveOnly=%b", routeId, groupInterval, groupActiveOnly);
+ }
+ };
+
+ return builders;
+ }
+
+ @Test
+ public void testLoopbackWithOneMessage() throws Exception {
+ String testMessage = generateMessage();
+ acknowledged.expectedBodiesReceived(testMessage);
+
+ String acknowledgement = source.requestBody((Object) testMessage, String.class);
+ Assert.assertThat("Should be acknowledgment for message 1", acknowledgement, CoreMatchers.containsString(String.format("MSA|AA|00001")));
+
+ assertMockEndpointsSatisfied(60, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testLoopbackMultipleMessages() throws Exception {
+ int messageCount = 1000;
+ acknowledged.expectedMessageCount(messageCount);
+
+ for (int i = 1; i <= messageCount; ++i) {
+ String testMessage = generateMessage(i);
+ acknowledged.message(i - 1).body().isEqualTo(testMessage);
+ String acknowledgement = source.requestBody((Object) testMessage, String.class);
+ Assert.assertThat("Should be acknowledgment for message " + i, acknowledgement, CoreMatchers.containsString(String.format("MSA|AA|%05d", i)));
+
+ }
+
+ assertMockEndpointsSatisfied(60, TimeUnit.SECONDS);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientConsumerBlueprintTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientConsumerBlueprintTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientConsumerBlueprintTest.java
new file mode 100644
index 0000000..e4ca285
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientConsumerBlueprintTest.java
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.blueprint.CamelBlueprintTestSupport;
+import org.junit.Ignore;
+
+@Ignore(value = "Not Yet Implemented")
+// TODO: Implement this
+public class MllpTcpClientConsumerBlueprintTest extends CamelBlueprintTestSupport {
+ @EndpointInject(uri = "mock://target")
+ MockEndpoint target;
+
+ @Override
+ protected String getBlueprintDescriptor() {
+ return "OSGI-INF/blueprint/mllp-tcp-client-consumer.xml";
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerAcknowledgementTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerAcknowledgementTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerAcknowledgementTest.java
new file mode 100644
index 0000000..af41550
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerAcknowledgementTest.java
@@ -0,0 +1,138 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit.rule.mllp.MllpServerResource;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+
+public class MllpTcpClientProducerAcknowledgementTest extends CamelTestSupport {
+ @Rule
+ public MllpServerResource mllpServer = new MllpServerResource(AvailablePortFinder.getNextAvailable());
+
+ @EndpointInject(uri = "direct://source")
+ ProducerTemplate source;
+
+ @EndpointInject(uri = "mock://complete")
+ MockEndpoint complete;
+
+ @EndpointInject(uri = "mock://aa-ack")
+ MockEndpoint accept;
+ @EndpointInject(uri = "mock://ae-nack")
+ MockEndpoint error;
+ @EndpointInject(uri = "mock://ar-nack")
+ MockEndpoint reject;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext) super.createCamelContext();
+
+ context.setUseMDCLogging(true);
+ context.setName(this.getClass().getSimpleName());
+
+ return context;
+ }
+
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ String routeId = "mllp-sender";
+
+ String host = "0.0.0.0";
+ int port = mllpServer.getListenPort();
+
+ public void configure() {
+ onException(MllpApplicationRejectAcknowledgementException.class)
+ .handled(true)
+ .to(reject)
+ .log(LoggingLevel.ERROR, routeId, "AR Acknowledgemnet");
+
+ onException(MllpApplicationErrorAcknowledgementException.class)
+ .handled(true)
+ .to(error)
+ .log(LoggingLevel.ERROR, routeId, "AE Acknowledgement");
+
+ onCompletion()
+ .onCompleteOnly()
+ .to(complete)
+ .log(LoggingLevel.DEBUG, routeId, "AA Acknowledgement");
+
+ from(source.getDefaultEndpoint()).routeId(routeId)
+ .log(LoggingLevel.INFO, routeId, "Sending Message")
+ .toF("mllp://%s:%d", host, port)
+ .log(LoggingLevel.INFO, routeId, "Received Acknowledgement")
+ .to(accept);
+ }
+ };
+ }
+
+ @Test
+ public void testApplicationAcceptAcknowledgement() throws Exception {
+ complete.setExpectedMessageCount(1);
+ accept.setExpectedMessageCount(1);
+ reject.setExpectedMessageCount(0);
+ error.setExpectedMessageCount(0);
+
+ source.sendBody(generateMessage());
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testApplicationRejectAcknowledgement() throws Exception {
+ complete.setExpectedMessageCount(1);
+ accept.setExpectedMessageCount(0);
+ reject.setExpectedMessageCount(1);
+ error.setExpectedMessageCount(0);
+
+ mllpServer.setSendApplicationRejectAcknowledgementModulus(1);
+
+ source.sendBody(generateMessage());
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testApplicationErrorAcknowledgement() throws Exception {
+ complete.setExpectedMessageCount(1);
+ accept.setExpectedMessageCount(0);
+ reject.setExpectedMessageCount(0);
+ error.setExpectedMessageCount(1);
+
+ mllpServer.setSendApplicationErrorAcknowledgementModulus(1);
+
+ source.sendBody(generateMessage());
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBlueprintTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBlueprintTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBlueprintTest.java
new file mode 100644
index 0000000..4233d80
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerBlueprintTest.java
@@ -0,0 +1,123 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.Dictionary;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultComponentResolver;
+import org.apache.camel.spi.ComponentResolver;
+import org.apache.camel.test.blueprint.CamelBlueprintTestSupport;
+import org.apache.camel.test.junit.rule.mllp.MllpServerResource;
+import org.apache.camel.util.KeyValueHolder;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+public class MllpTcpClientProducerBlueprintTest extends CamelBlueprintTestSupport {
+ @Rule
+ public MllpServerResource mllpServer = new MllpServerResource();
+
+ final String sourceUri = "direct://source";
+ final String mockAcknowledgedUri = "mock://acknowledged";
+ final String mockTimeoutUri = "mock://timeout-ex";
+ final String mockAeExUri = "mock://ae-ack";
+ final String mockArExUri = "mock://ar-ack";
+ final String mockFrameExUri = "mock://frame-ex";
+
+ @EndpointInject(uri = sourceUri)
+ ProducerTemplate source;
+ @EndpointInject(uri = mockAcknowledgedUri)
+ MockEndpoint acknowledged;
+ @EndpointInject(uri = mockTimeoutUri)
+ MockEndpoint timeout;
+ @EndpointInject(uri = mockAeExUri)
+ MockEndpoint ae;
+ @EndpointInject(uri = mockArExUri)
+ MockEndpoint ar;
+ @EndpointInject(uri = mockFrameExUri)
+ MockEndpoint frame;
+
+ @Override
+ protected String getBlueprintDescriptor() {
+ return "OSGI-INF/blueprint/mllp-tcp-client-producer-test.xml";
+ }
+
+ @Override
+ protected Properties useOverridePropertiesWithPropertiesComponent() {
+ Properties props = new Properties();
+
+ props.setProperty("sourceUri", sourceUri);
+ props.setProperty("acknowledgedUri", mockAcknowledgedUri);
+ props.setProperty("timeoutUri", mockTimeoutUri);
+ props.setProperty("frameErrorUri", mockFrameExUri);
+ props.setProperty("errorAcknowledgementUri", mockAeExUri);
+ props.setProperty("rejectAcknowledgementUri", mockArExUri);
+
+ props.setProperty("mllp.port", Integer.toString(mllpServer.getListenPort()));
+
+ return props;
+ }
+
+ /*
+ This doesn't seem to work
+ @Override
+ protected String useOverridePropertiesWithConfigAdmin(Dictionary props) throws Exception {
+
+ props.put("mllp.port", mllpServer.getListenPort() );
+
+ return "MllpTcpClientProducer";
+ }
+ */
+
+ @Override
+ protected void addServicesOnStartup(Map<String, KeyValueHolder<Object, Dictionary>> services) {
+ ComponentResolver testResolver = new DefaultComponentResolver();
+
+ services.put(ComponentResolver.class.getName(), asService(testResolver, "component", "mllp"));
+ }
+
+ @Test()
+ public void testSendMultipleMessages() throws Exception {
+ int messageCount = 500;
+ acknowledged.setExpectedMessageCount(messageCount);
+ timeout.setExpectedMessageCount(0);
+ frame.setExpectedMessageCount(0);
+ ae.setExpectedMessageCount(0);
+ ar.setExpectedMessageCount(0);
+
+ // Uncomment one of these lines to see the NACKs handled
+ // mllpServer.setSendApplicationRejectAcknowledgementModulus(10);
+ // mllpServer.setSendApplicationErrorAcknowledgementModulus(10);
+
+ for (int i = 0; i < messageCount; ++i) {
+ log.debug("Triggering message {}", i);
+ Object response = source.requestBodyAndHeader(generateMessage(i), "CamelMllpMessageControlId", String.format("%05d", i));
+ log.debug("response {}\n{}", i, response);
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+
+}
[4/6] camel git commit: Added camel-mllp component
Posted by da...@apache.org.
Added camel-mllp component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/bd1661b2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/bd1661b2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/bd1661b2
Branch: refs/heads/master
Commit: bd1661b248c960e200bd5f69e1ba34989529a712
Parents: c34db42
Author: Quinn Stevenson <qu...@pronoia-solutions.com>
Authored: Mon Dec 28 11:38:14 2015 -0700
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Dec 30 09:43:10 2015 +0100
----------------------------------------------------------------------
.../src/main/descriptors/common-bin.xml | 1 +
components/camel-mllp/LIMITATIONS.md | 35 +
components/camel-mllp/README.md | 132 +++
components/camel-mllp/pom.xml | 101 +++
.../mllp/MllpAcknowledgementException.java | 39 +
.../MllpAcknowledgementTimoutException.java | 38 +
...pplicationErrorAcknowledgementException.java | 38 +
...plicationRejectAcknowledgementException.java | 38 +
.../camel/component/mllp/MllpComponent.java | 49 ++
.../camel/component/mllp/MllpConstants.java | 64 ++
.../mllp/MllpCorruptFrameException.java | 40 +
.../camel/component/mllp/MllpEndpoint.java | 283 ++++++
.../camel/component/mllp/MllpException.java | 49 ++
.../MllpInvalidAcknowledgementException.java | 38 +
.../component/mllp/MllpTcpClientProducer.java | 232 +++++
.../component/mllp/MllpTcpServerConsumer.java | 550 ++++++++++++
.../component/mllp/MllpTimeoutException.java | 39 +
.../component/mllp/MllpWriteException.java | 39 +
.../camel/component/mllp/impl/MllpUtil.java | 371 ++++++++
.../Hl7AcknowledgementGenerationException.java | 41 +
.../mllp/Hl7AcknowledgementGenerator.java | 150 ++++
.../services/org/apache/camel/component/mllp | 1 +
.../mllp/MllpProducerConsumerLoopbackTest.java | 121 +++
.../MllpTcpClientConsumerBlueprintTest.java | 36 +
...llpTcpClientProducerAcknowledgementTest.java | 138 +++
.../MllpTcpClientProducerBlueprintTest.java | 123 +++
.../mllp/MllpTcpClientProducerTest.java | 249 ++++++
.../MllpTcpServerConsumerBlueprintTest.java | 101 +++
.../MllpTcpServerConsumerConnectionTest.java | 102 +++
...pTcpServerConsumerMulitpleTcpPacketTest.java | 120 +++
.../mllp/MllpTcpServerConsumerTest.java | 169 ++++
.../MllpTcpServerProducerBlueprintTest.java | 35 +
.../junit/rule/mllp/MllpClientResource.java | 461 ++++++++++
.../MllpJUnitResourceCorruptFrameException.java | 38 +
.../rule/mllp/MllpJUnitResourceException.java | 38 +
.../mllp/MllpJUnitResourceTimeoutException.java | 38 +
.../junit/rule/mllp/MllpServerResource.java | 878 +++++++++++++++++++
.../camel/test/mllp/Hl7MessageGenerator.java | 84 ++
.../camel/test/mllp/PassthroughProcessor.java | 47 +
.../apache/camel/test/tcp/JavaSocketTests.java | 176 ++++
.../blueprint/mllp-tcp-client-producer-test.xml | 114 +++
.../blueprint/mllp-tcp-server-consumer-test.xml | 44 +
.../src/test/resources/log4j.properties | 41 +
components/pom.xml | 1 +
parent/pom.xml | 5 +
.../features/src/main/resources/features.xml | 3 +
46 files changed, 5530 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/apache-camel/src/main/descriptors/common-bin.xml
----------------------------------------------------------------------
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index 9d2e447..cd70770 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -142,6 +142,7 @@
<include>org.apache.camel:camel-mail</include>
<include>org.apache.camel:camel-mina</include>
<include>org.apache.camel:camel-mina2</include>
+ <include>org.apache.camel:camel-mllp</include>
<include>org.apache.camel:camel-mongodb</include>
<include>org.apache.camel:camel-mqtt</include>
<include>org.apache.camel:camel-msv</include>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/LIMITATIONS.md
----------------------------------------------------------------------
diff --git a/components/camel-mllp/LIMITATIONS.md b/components/camel-mllp/LIMITATIONS.md
new file mode 100644
index 0000000..dab7957
--- /dev/null
+++ b/components/camel-mllp/LIMITATIONS.md
@@ -0,0 +1,35 @@
+Does not support CA, CE and CR Acknowledgements (no plans to implement)
+
+The MLLP Sequence protocol is not implemented (no plans to implement)
+
+The MllpTcpClientConsumer is not implemented yet
+
+The MllpTcpServerProducer is not implemented yet
+
+idleTimeout is not enforced yet - connections can remain open indefinately
+
+maxConnections are not enforced yet - an unlimited number of connections are allowed
+
+Suspending a MLLP Consumer is not yet implemented
+
+RouteBuilders for standard usage -
+ The plan for this is to have a few RouteBuilders packaged with the component that
+ can be used as-is for some of the standard use-cases ( i.e. MLLP to JMS, JMS to MLLP, etc).
+
+JMX Instrumentation is not yet implemented
+ - Manually reset connection
+ - Suspend a consumer
+
+MLLP Statistics are not yet implemented
+ TCP Server
+ - last message received time
+ - last AA sent time
+ - last AE sent time
+ - last AR sent time
+ - AA, AE & AR counts
+ - last message sent time
+ - last message sent time
+ - last AA received time
+ - last AE received time
+ - last AR received time
+ - AA, AE & AR counts
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/README.md
----------------------------------------------------------------------
diff --git a/components/camel-mllp/README.md b/components/camel-mllp/README.md
new file mode 100644
index 0000000..e7b0fc9
--- /dev/null
+++ b/components/camel-mllp/README.md
@@ -0,0 +1,132 @@
+# camel-mllp
+Camel MLLP Component
+
+See MLLP Specification here:
+http://www.hl7.org/documentcenter/public_temp_E7494E36-1C23-BA17-0C5E72EF77542E1F/wg/inm/mllp_transport_specification.PDF
+
+# Description
+The camel-mllp component is specifically designed to handle the nuances of the MLLP protocol and provide
+the functionality required by Healthcare providers to communicate with other systems using the MLLP protocol.
+
+The component is capable of handling both byte[] and String payloads which allows using other HL7 Libraries (i.e. HAPI)
+to parse the messages. When String payloads are specified, the component will use the systems default Charset for
+encoding/decoding, but this can be overridden.
+
+The component provides a simple URI for configuring MLLP endpoints:
+MLLP-Producers:
+ to( "mllp://<host or ip>:<port>" )
+MLLP-Consumers:
+ from("mllp://<host or ip>:<listening port>) to listen on a specific address
+ from("mllp://port") to listen on all local addresses
+
+MLLP-Producers also interrogate the HL7 Acknowledgment received from the external system and if a negative acknowledgment
+is received, the producer sets an exception on the exchange indicating the type of negative acknowledgement that was
+received (i.e. a HL7 Application Reject Acknowledgement or a HL7 Application Error Acknowledgement). This enables
+the use of Camel Redelivery Policies to configure redelivery attempts and routing erroneous messages to alternate
+endpoints for analysis.
+
+MLLP-Consumers will, by default, automatically generate an acknowledgement. A HL7 Application Accept Acknowledgment
+will be generated for successfully processed messages, or a HL7 Application Error Acknowledgement for messages where an
+exception is raised during the processing of the exchange. The HL7 acknowledgement can also be specified by setting the
+CamelMllpAcknowledgement property on the exchange - if present, the value of this property will be used for the HL7
+acknowledgment. The automatic generation of an HL7 acknowledgment can be completely disabled by setting the autoAck
+URI parameter to false. If autoAck=false and the CamelMllpAcknowledgment property is not set on the exchange, and
+exception will be raised.
+
+The component also provides a Camel Processor that is capable of generating HL7 Acknowledgements. Therefore, the HAPI
+is not required to generate HL7 Acknowledgements - however, it can be used if desired.
+By default, the processor will generate a HL7 Application Accept Acknowledgement if there is not an exception on the
+Camel Exchange - otherwise it will generate a HL7 Application Error Acknowledgement. The generated acknowledgement
+is placed in the CamelMllpAcknowledgment property on the Exchange.
+
+Regardless of whether the HL7 Acknowledgment is generated or specified using the CamelMllpAcknowledgement Exchange property,
+the MLLP-Consumer will set the CamelHllpAcknowledgement and CamelHllpAcknowledgementCode headers on the Message after the
+acknowledgment is successfully transmitted to the external system.
+
+Since the MLLP protocol does not typically use a large number of concurrent connections, the camel-mllp component uses
+a simple thread-per-connection model based an standard Java Sockets. This keeps the implementation simple, and also
+eliminates the dependencies on other camel components.
+
+# Rationalization
+While Camel already includes some of this functionality in the camel-hl7 component, the implementation does not
+provide all of the functionality required to effectively handle the MLLP protocol - especially under adverse conditions.
+
+The camel-hl7 Mina2 codec and Netty4 decoder do not handle MLLP Framing errors very well - the component will hang waiting
+for frames to complete in some instances.
+
+While both camel-mina2 and camel-netty4 provide a "timeout" function, it is only applied to Producers. MLLP Consumers
+also need to be able to timeout to recover from MLLP framing errors. Additionally, the timeout functionality of the
+camel-netty4 component is disable after any data is received on the connection, making in ineffective for detecting
+timeouts after the first messages is received.
+
+Also, neither the Mina2 codec nor the Netty4 decoder interrogate HL7 Acknowledgments. Therefore, it is much more
+difficult to use the redelivery and error handling features Camel provides.
+
+The above issues may be addressable by updating/patching the existing components, but there is one more that is not.
+Both camel-netty4 and camel-mina2 are designed to handle a large number of concurrent connections, rapid connect/disconnect
+rates, and asynchronous communication. Forcing, these components to deal with the small number of stateful connections
+inherent to the MLLP protocol seems inappropriate.
+
+An attempt was made to update the camel-netty4 decoder provided by the camel-hl7 component to deal with the nuances of
+the MLLP protocol, but it quickly became very complicated. The decoder was updated to correctly deal with the MLLP
+frame. The current implementation is based on the Netty DelimiterBasedFrameDecoder, but this decoder only looks for a
+single byte terminator and MLLP uses two bytes to terminate it's frame. Additionally, the second terminating byte of the
+MLLP frame can't be used alone because that byte is also contained in HL7 messages (it is the HL7 Segment Delimiter).
+An implementation resembling the Netty LineBasedFrameDecoder was written at it correctly handled the MLLP framing issues,
+but the timout issues were never addressed.
+
+For MLLP Consumers, the decoder needs to dynamically install a timeout handler whenver a partial MLLP frame is received,
+and then remove it when the frame is completed. For MLLP Producers, the decoder would need to install a timeout handler
+at some point to enable detecting a missing acknowledgement in addition to the timeout handler to deal with the
+partial/incomplete acknowledgement.
+
+# MLLP Background
+The MLLP protocol is inherently synchronous because external systems almost always require the order of messages to be
+maintained (i.e. FIFO delivery).
+
+When a MLLP-Producer sends a message to an external system, it is required to wait for
+an HL7 Acknowledgement before sending the next message. Additionally, the content of the acknowlegement must be examined
+to determine the specific type of the acknowlegement before the next message can be transmitted. If the acknowledgement
+is a HL7 Application Error Acknowledgement, the MLLP-Producer should retransmit the message a few times (the number of
+redelivery attempts is application specific). If the acknowledgment is a HL7 Application Reject Acknowledgement, there
+is something wrong with the message and redelivery will never succeed. The HL7 Messages acknowledged with an HL7 Application
+Reject Acknowledgement must be routed to an alternate destination to allow users to investigate the nature of the error
+so it can be corrected.
+
+When a MLLP-Consumer receives a message from an external system, the sending system is required to wait for an HL7 Acknowledgement
+before transmitting the next message. The MLLP-Consumer normally persists the message in a durable store and then replies
+to the sending system with a HL7 Application Accept Acknowledgement. If a transient error occurs while persisting the
+message, the MLLP-Consumer should reply with a HL7 Application Error Acknowledgment and allow the external system to resend
+the message. If the MLLP-Consumer detects the that the received message is invalid for some reason and the message could
+never be processed, it should reply with an HL7 Application Reject Acknowledgement and the sending system should not
+attempt to resend the message.
+
+NOTE: Some external systems do not handle HL7 NACKS ( HL7 Application Reject Acknowledgments and HL7 Application Error
+Acknowledgements) - they do not interrogate the HL7 Acknowledgment to determine if it is a negative acknowledgement and
+assume any acknowledgement received is an HL7 ACK (HL7 Application Accept Acknowledgement). In order to prevent message
+loss when dealing with external systems that behave in this fashion, the MLLP-Consumer must be capable of closing the
+TCP connection in lew of sending and HL7 NACK, which will force the external system to resend the message. Additionally,
+the MLLP-Consumer may be required to behave differently for each type of HL7 NACK - it may need to close the TCP connection
+instead of sending HL7 Application Error Acknowledgements, and route the messages that would be normally not be persisted
+to an alternate durable store before sending the the HL7 Application Reject acknowledgement.
+
+Systems using the MLLP protocol normally use stateful TCP connections - the connections are established and left open
+for extended periods of time.
+
+A MLLP-Consumer endpoint may have more than one TCP connection at a given time, but this is not the typical case. Normally
+there is a single active TCP connection to a MLLP-Consumer.
+
+A MLLP-Producer endpoint should only have a single TCP connection at any given time. If the producer attempts to open more
+than one connection to an external system, it oftentimes causes issues with the external system. Additionally, since FIFO
+must be maintained, only
+
+# CamelPatches
+
+To help simplify bringing Camel-MLLP into Apache Camel, we have take care to build a set of patches that help integrate the component
+into Camel.
+
+features.xml is a diff file that will add Camel-MLLP to the camel features list when on Apache Karaf.
+
+parent-pom.xml is a diff file that contains an update to Camel's Parent pom.
+
+common-bin.xml is a diff file that contains an update to the common distribution's include statements.
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-mllp/pom.xml b/components/camel-mllp/pom.xml
new file mode 100644
index 0000000..3a41bcf
--- /dev/null
+++ b/components/camel-mllp/pom.xml
@@ -0,0 +1,101 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.17-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-mllp</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: MLLP</name>
+ <description>Camel MLLP Component</description>
+
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.mllp.*,org.apache.camel.processor.mllp.*</camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=mllp</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+
+ <!-- support camel documentation -->
+ <!--dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>apt</artifactId>
+ </dependency -->
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+
+ <!-- testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-blueprint</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- test logging -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <!-- Ensure that test runs do no leave running JVMs -->
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.19</version>
+ <configuration>
+ <shutdown>kill</shutdown>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementException.java
new file mode 100644
index 0000000..624745c
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementException.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Base class for HL7 Application Acknowledgement Exceptions
+ */
+public abstract class MllpAcknowledgementException extends MllpException {
+ public MllpAcknowledgementException(String message) {
+ super(message);
+ }
+
+ public MllpAcknowledgementException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpAcknowledgementException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpAcknowledgementException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementTimoutException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementTimoutException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementTimoutException.java
new file mode 100644
index 0000000..9288a14
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpAcknowledgementTimoutException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer does not receive a HL7 acknowledgement within the configured timespan
+ */
+public class MllpAcknowledgementTimoutException extends MllpTimeoutException {
+ public MllpAcknowledgementTimoutException(String message) {
+ super(message);
+ }
+
+ public MllpAcknowledgementTimoutException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpAcknowledgementTimoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpAcknowledgementTimoutException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationErrorAcknowledgementException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationErrorAcknowledgementException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationErrorAcknowledgementException.java
new file mode 100644
index 0000000..7d7ecd2
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationErrorAcknowledgementException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer receives a HL7 Application Error Acknowledgement
+ */
+public class MllpApplicationErrorAcknowledgementException extends MllpAcknowledgementException {
+ public MllpApplicationErrorAcknowledgementException(String message) {
+ super(message);
+ }
+
+ public MllpApplicationErrorAcknowledgementException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpApplicationErrorAcknowledgementException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpApplicationErrorAcknowledgementException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationRejectAcknowledgementException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationRejectAcknowledgementException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationRejectAcknowledgementException.java
new file mode 100644
index 0000000..ef2a39d
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpApplicationRejectAcknowledgementException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer receives a HL7 Application Reject Acknowledgement
+ */
+public class MllpApplicationRejectAcknowledgementException extends MllpAcknowledgementException {
+ public MllpApplicationRejectAcknowledgementException(String message) {
+ super(message);
+ }
+
+ public MllpApplicationRejectAcknowledgementException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpApplicationRejectAcknowledgementException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpApplicationRejectAcknowledgementException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
new file mode 100644
index 0000000..d815158
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpComponent.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Represents the component that manages {@link MllpEndpoint}.
+ */
+public class MllpComponent extends UriEndpointComponent {
+ public static final String MLLP_LOG_PHI_PROPERTY = "org.apache.camel.mllp.logPHI";
+
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ public MllpComponent() {
+ super(MllpEndpoint.class);
+ }
+
+ public MllpComponent(CamelContext context) {
+ super(context, MllpEndpoint.class);
+ }
+
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ Endpoint endpoint = new MllpEndpoint(uri, this);
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpConstants.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpConstants.java
new file mode 100644
index 0000000..da5eb77
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpConstants.java
@@ -0,0 +1,64 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+public final class MllpConstants {
+ public static final String MLLP_LOCAL_ADDRESS = "CamelMllpLocalAddress";
+ public static final String MLLP_REMOTE_ADDRESS = "CamelMllpRemoteAddress";
+
+ public static final String MLLP_ACKNOWLEDGEMENT = "CamelMllpAcknowledgement";
+ public static final String MLLP_ACKNOWLEDGEMENT_TYPE = "CamelMllpAcknowledgementType";
+
+ /*
+ Connection Control Exchange Properties
+ - For Consumers, "SEND" => ACKNOWLEDGEMENT
+ - For Producers, "SEND" => MESSAGE
+ */
+ public static final String MLLP_CLOSE_CONNECTION_BEFORE_SEND = "CamelMllpCloseConnectionBeforeSend";
+ public static final String MLLP_RESET_CONNECTION_BEFORE_SEND = "CamelMllpResetConnectionBeforeSend";
+ public static final String MLLP_CLOSE_CONNECTION_AFTER_SEND = "CamelMllpCloseConnectionAfterSend";
+ public static final String MLLP_RESET_CONNECTION_AFTER_SEND = "CamelMllpResetConnectionAfterSend";
+
+ // MSH-3
+ public static final String MLLP_SENDING_APPLICATION = "CamelMllpSendingApplication";
+ // MSH-4
+ public static final String MLLP_SENDING_FACILITY = "CamelMllpSendingFacility";
+ // MSH-5
+ public static final String MLLP_RECEIVING_APPLICATION = "CamelMllpReceivingApplication";
+ // MSH-6
+ public static final String MLLP_RECEIVING_FACILITY = "CamelMllpReceivingFacility";
+ // MSH-7
+ public static final String MLLP_TIMESTAMP = "CamelMllpTimestamp";
+ // MSH-8
+ public static final String MLLP_SECURITY = "CamelMllpSecurity";
+ // MSH-9
+ public static final String MLLP_MESSAGE_TYPE = "CamelMllpMessageType";
+ // MSH-9.1
+ public static final String MLLP_EVENT_TYPE = "CamelMllpEventType";
+ // MSH-9.2
+ public static final String MLLP_TRIGGER_EVENT = "CamelMllpTriggerEvent";
+ // MSH-10
+ public static final String MLLP_MESSAGE_CONTROL = "CamelMllpMessageControlId";
+ // MSH-11
+ public static final String MLLP_PROCESSING_ID = "CamelMllpProcessingId";
+ // MSH-12
+ public static final String MLLP_VERSION_ID = "CamelMllpVersionId";
+ // MSH-18
+ public static final String MLLP_CHARSET = "CamelMllpCharset";
+
+ private MllpConstants() { }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpCorruptFrameException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpCorruptFrameException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpCorruptFrameException.java
new file mode 100644
index 0000000..0310530
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpCorruptFrameException.java
@@ -0,0 +1,40 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer or consumer encounters a corrupt MLLP Frame while attempting
+ * to read or write a MLLP payload.
+ */
+public class MllpCorruptFrameException extends MllpException {
+ public MllpCorruptFrameException(String message) {
+ super(message);
+ }
+
+ public MllpCorruptFrameException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpCorruptFrameException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpCorruptFrameException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
new file mode 100644
index 0000000..eda5757
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpEndpoint.java
@@ -0,0 +1,283 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Represents a MLLP endpoint.
+ *
+ * NOTE: MLLP payloads are not logged unless the logging level is set to DEBUG or TRACE to avoid introducing PHI
+ * into the log files. Logging of PHI can be globally disabled by setting the org.apache.camel.mllp.logPHI system
+ * property to false.
+ * <p/>
+ */
+@UriEndpoint(scheme = "mllp", title = "mllp", syntax = "mllp:hostname:port", consumerClass = MllpTcpServerConsumer.class, label = "mllp")
+public class MllpEndpoint extends DefaultEndpoint {
+ public static final char START_OF_BLOCK = 0x0b; // VT (vertical tab) - decimal 11, octal 013
+ public static final char END_OF_BLOCK = 0x1c; // FS (file separator) - decimal 28, octal 034
+ public static final char END_OF_DATA = 0x0d; // CR (carriage return) - decimal 13, octal 015
+ public static final int END_OF_STREAM = -1; //
+ public static final char SEGMENT_DELIMITER = 0x0d; // CR (carriage return) - decimal 13, octal 015
+ public static final char MESSAGE_TERMINATOR = 0x0a; // LF (line feed, new line) - decimal 10, octal 012
+
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ @UriPath(defaultValue = "0.0.0.0", description = "Hostname or IP for connection")
+ String hostname = "0.0.0.0";
+
+ @UriPath(description = "TCP Port for connection")
+ int port = -1;
+
+ // TODO: Move URI Params to a MllpConfiguration class
+ @UriParam(defaultValue = "5", description = "TCP Server only - The maximum queue length for incoming connection indications (a request to connect) is set to the backlog parameter. If a "
+ + "connection indication arrives when the queue is full, the connection is refused.")
+ int backlog = 5;
+
+ @UriParam(defaultValue = "30000", description = "TCP Server only - timeout value while waiting for a TCP listener to start (milliseconds)")
+ int bindTimeout = 30000;
+
+ @UriParam(defaultValue = "30000", description = "TCP Server only - timeout value while waiting for a TCP connection (milliseconds)")
+ int acceptTimeout = 30000;
+
+ @UriParam(defaultValue = "30000", description = "TCP Client only - timeout value while establishing for a TCP connection (milliseconds)")
+ int connectTimeout = 30000;
+
+ @UriParam(defaultValue = "5000", description = "Timeout value (milliseconds) used when reading a message from an external")
+ int responseTimeout = 5000;
+
+ @UriParam(defaultValue = "true", description = "Enable/disable the SO_KEEPALIVE socket option.")
+ boolean keepAlive = true;
+
+ @UriParam(defaultValue = "true", description = "Enable/disable the TCP_NODELAY socket option.")
+ boolean tcpNoDelay = true;
+
+ @UriParam(defaultValue = "false", description = "Enable/disable the SO_REUSEADDR socket option.")
+ boolean reuseAddress;
+
+ @UriParam(description = "Sets the SO_RCVBUF option to the specified value")
+ Integer receiveBufferSize;
+
+ @UriParam(description = "Sets the SO_SNDBUF option to the specified value")
+ Integer sendBufferSize;
+
+ @UriParam(defaultValue = "0", description = "The amount of time a TCP connection can remain idle before it is closed")
+ int idleTimeout;
+
+ @UriParam(description = "The TCP mode of the endpoint (client or server). Defaults to client for Producers and server for Consumers")
+ String tcpMode;
+
+ @UriParam(defaultValue = "true", description = "MLLP Consumers only - Automatically generate and send an MLLP Acknowledgement")
+ boolean autoAck = true;
+
+ @UriParam(description = "Set the CamelCharsetName property on the exchange")
+ String charsetName;
+
+ public MllpEndpoint(String uri, MllpComponent component) {
+ super(uri, component);
+
+ // mllp://hostname:port
+ String hostPort;
+ // look for options
+ int optionsStartIndex = uri.indexOf('?');
+ if (-1 == optionsStartIndex) {
+ // No options - just get the host/port stuff
+ hostPort = uri.substring(7);
+ } else {
+ hostPort = uri.substring(7, optionsStartIndex);
+ }
+
+ // Make sure it has a host - may just be a port
+ int colonIndex = hostPort.indexOf(':');
+ if (-1 != colonIndex) {
+ hostname = hostPort.substring(0, colonIndex);
+ port = Integer.parseInt(hostPort.substring(colonIndex + 1));
+ } else {
+ // No host specified - leave the default host and set the port
+ port = Integer.parseInt(hostPort);
+ }
+ }
+
+
+ @Override
+ public ExchangePattern getExchangePattern() {
+ return ExchangePattern.InOut;
+ }
+
+ @Override
+ public Exchange createExchange() {
+ return this.createExchange(getExchangePattern());
+ }
+
+ @Override
+ public Exchange createExchange(ExchangePattern exchangePattern) {
+ Exchange mllpExchange = super.createExchange(exchangePattern);
+ setExchangeProperties(mllpExchange);
+
+ return mllpExchange;
+ }
+
+ @Override
+ public Exchange createExchange(Exchange exchange) {
+ Exchange mllpExchange = super.createExchange(exchange);
+ setExchangeProperties(mllpExchange);
+
+ return mllpExchange;
+ }
+
+ private void setExchangeProperties(Exchange mllpExchange) {
+ if (null != charsetName) {
+ mllpExchange.setProperty(Exchange.CHARSET_NAME, charsetName);
+ }
+ }
+
+ public Producer createProducer() throws Exception {
+ log.trace("({}).createProducer()", this.getEndpointKey());
+
+ return new MllpTcpClientProducer(this);
+ }
+
+ public Consumer createConsumer(Processor processor) throws Exception {
+ log.trace("({}).createConsumer(processor)", this.getEndpointKey());
+
+ return new MllpTcpServerConsumer(this, processor);
+ }
+
+ @Override
+ public boolean isSynchronous() {
+ return true;
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public String getCharsetName() {
+ return charsetName;
+ }
+
+ public void setCharsetName(String charsetName) {
+ this.charsetName = charsetName;
+ }
+
+ public String getHostname() {
+ return hostname;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ /**
+ * The maximum queue length for incoming connection indications (a request to connect) is set to the backlog parameter. If a connection indication arrives when the queue is full, the connection
+ * is refused.
+ */
+ public int getBacklog() {
+ return backlog;
+ }
+
+ /**
+ * The maximum queue length for incoming connection indications (a request to connect) is set to the backlog parameter. If a connection indication arrives when the queue is full, the connection
+ * is refused.
+ */
+ public void setBacklog(int backlog) {
+ this.backlog = backlog;
+ }
+
+ public int getAcceptTimeout() {
+ return acceptTimeout;
+ }
+
+ public void setAcceptTimeout(int acceptTimeout) {
+ this.acceptTimeout = acceptTimeout;
+ }
+
+ public int getConnectTimeout() {
+ return connectTimeout;
+ }
+
+ public void setConnectTimeout(int connectTimeout) {
+ this.connectTimeout = connectTimeout;
+ }
+
+ public int getResponseTimeout() {
+ return responseTimeout;
+ }
+
+ public void setResponseTimeout(int responseTimeout) {
+ this.responseTimeout = responseTimeout;
+ }
+
+ public boolean isKeepAlive() {
+ return keepAlive;
+ }
+
+ public void setKeepAlive(boolean keepAlive) {
+ this.keepAlive = keepAlive;
+ }
+
+ public boolean isTcpNoDelay() {
+ return tcpNoDelay;
+ }
+
+ public void setTcpNoDelay(boolean tcpNoDelay) {
+ this.tcpNoDelay = tcpNoDelay;
+ }
+
+ public boolean isReuseAddress() {
+ return reuseAddress;
+ }
+
+ public void setReuseAddress(boolean reuseAddress) {
+ this.reuseAddress = reuseAddress;
+ }
+
+ public int getReceiveBufferSize() {
+ return receiveBufferSize;
+ }
+
+ public void setReceiveBufferSize(int receiveBufferSize) {
+ this.receiveBufferSize = receiveBufferSize;
+ }
+
+ public int getSendBufferSize() {
+ return sendBufferSize;
+ }
+
+ public void setSendBufferSize(int sendBufferSize) {
+ this.sendBufferSize = sendBufferSize;
+ }
+
+ public boolean isAutoAck() {
+ return autoAck;
+ }
+
+ public void setAutoAck(boolean autoAck) {
+ this.autoAck = autoAck;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpException.java
new file mode 100644
index 0000000..a1bb182
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpException.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Base class for all MLLP Exceptions, and also used as a generic MLLP exception
+ */
+public class MllpException extends Exception {
+ private final byte[] mllpPayload;
+
+ public MllpException(String message) {
+ super(message);
+ this.mllpPayload = null;
+ }
+
+ public MllpException(String message, byte[] mllpPayload) {
+ super(message);
+ this.mllpPayload = mllpPayload;
+ }
+
+ public MllpException(String message, Throwable cause) {
+ super(message, cause);
+ this.mllpPayload = null;
+ }
+
+ public MllpException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, cause);
+ this.mllpPayload = mllpPayload;
+ }
+
+ public byte[] getMllpPayload() {
+ return mllpPayload;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpInvalidAcknowledgementException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpInvalidAcknowledgementException.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpInvalidAcknowledgementException.java
new file mode 100644
index 0000000..495cfae
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpInvalidAcknowledgementException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+/**
+ * Raised when a MLLP Producer receives a HL7 Acknowledgement for which the HL7 Acknowledgement type cannot be determined.
+ */
+public class MllpInvalidAcknowledgementException extends MllpAcknowledgementException {
+ public MllpInvalidAcknowledgementException(String message) {
+ super(message);
+ }
+
+ public MllpInvalidAcknowledgementException(String message, byte[] mllpPayload) {
+ super(message, mllpPayload);
+ }
+
+ public MllpInvalidAcknowledgementException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpInvalidAcknowledgementException(String message, byte[] mllpPayload, Throwable cause) {
+ super(message, mllpPayload, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
new file mode 100644
index 0000000..7b9d86b
--- /dev/null
+++ b/components/camel-mllp/src/main/java/org/apache/camel/component/mllp/MllpTcpClientProducer.java
@@ -0,0 +1,232 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.mllp.impl.MllpUtil;
+import org.apache.camel.impl.DefaultProducer;
+
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_ACKNOWLEDGEMENT;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_ACKNOWLEDGEMENT_TYPE;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_CLOSE_CONNECTION_AFTER_SEND;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_CLOSE_CONNECTION_BEFORE_SEND;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_LOCAL_ADDRESS;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_REMOTE_ADDRESS;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_RESET_CONNECTION_AFTER_SEND;
+import static org.apache.camel.component.mllp.MllpConstants.MLLP_RESET_CONNECTION_BEFORE_SEND;
+import static org.apache.camel.component.mllp.MllpEndpoint.SEGMENT_DELIMITER;
+
+/**
+ * The MLLP producer.
+ */
+public class MllpTcpClientProducer extends DefaultProducer {
+ MllpEndpoint endpoint;
+
+ Socket socket;
+
+ public MllpTcpClientProducer(MllpEndpoint endpoint) throws SocketException {
+ super(endpoint);
+ log.trace("MllpTcpClientProducer(endpoint)");
+
+ this.endpoint = endpoint;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ log.trace("doStart()");
+
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ log.trace("doStop()");
+
+ MllpUtil.closeConnection(socket);
+
+ super.doStop();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ log.trace("process(exchange)");
+
+ // Check BEFORE_SEND Properties
+ if (exchange.getProperty(MLLP_RESET_CONNECTION_BEFORE_SEND, boolean.class)) {
+ MllpUtil.resetConnection(socket);
+ return;
+ } else if (exchange.getProperty(MLLP_CLOSE_CONNECTION_BEFORE_SEND, boolean.class)) {
+ MllpUtil.closeConnection(socket);
+ }
+
+ Exception connectionException = checkConnection();
+ if (null != connectionException) {
+ exchange.setException(connectionException);
+ return;
+ }
+
+ Message message;
+ if (exchange.hasOut()) {
+ message = exchange.getOut();
+ } else {
+ message = exchange.getIn();
+ }
+
+ byte[] hl7MessageBytes = message.getMandatoryBody(byte[].class);
+
+ log.debug("Sending message to external system");
+ try {
+ MllpUtil.writeFramedPayload(socket, hl7MessageBytes);
+ } catch (MllpException mllpEx) {
+ exchange.setException(mllpEx);
+ return;
+ }
+
+ log.debug("Reading acknowledgement from external system");
+ byte[] acknowledgementBytes = null;
+ try {
+ MllpUtil.openFrame(socket);
+ acknowledgementBytes = MllpUtil.closeFrame(socket);
+ } catch (SocketTimeoutException timeoutEx) {
+ exchange.setException(new MllpAcknowledgementTimoutException("Acknowledgement timout", timeoutEx));
+ return;
+ } catch (MllpException mllpEx) {
+ exchange.setException(mllpEx);
+ return;
+ }
+
+ log.debug("Populating the exchange with the acknowledgement from the external system");
+ message.setHeader(MLLP_ACKNOWLEDGEMENT, acknowledgementBytes);
+
+ message.setHeader(MLLP_LOCAL_ADDRESS, socket.getLocalAddress().toString());
+ message.setHeader(MLLP_REMOTE_ADDRESS, socket.getRemoteSocketAddress());
+
+ // Now, extract the acknowledgement type and check for a NACK
+ byte fieldDelim = acknowledgementBytes[3];
+ // First, find the beginning of the MSA segment - should be the second segment
+ int msaStartIndex = -1;
+ for (int i = 0; i < acknowledgementBytes.length; ++i) {
+ if (SEGMENT_DELIMITER == acknowledgementBytes[i]) {
+ final byte bM = 77;
+ final byte bS = 83;
+ final byte bA = 65;
+ final byte bE = 69;
+ final byte bR = 82;
+ /* We've found the start of a new segment - make sure peeking ahead
+ won't run off the end of the array - we need at least 7 more bytes
+ */
+ if (acknowledgementBytes.length > i + 7) {
+ // We can safely peek ahead
+ if (bM == acknowledgementBytes[i + 1] && bS == acknowledgementBytes[i + 2] && bA == acknowledgementBytes[i + 3] && fieldDelim == acknowledgementBytes[i + 4]) {
+ // Found the beginning of the MSA - the next two bytes should be our acknowledgement code
+ msaStartIndex = i + 1;
+ if (bA != acknowledgementBytes[i + 5]) {
+ exchange.setException(new MllpInvalidAcknowledgementException(new String(acknowledgementBytes)));
+ } else {
+ switch (acknowledgementBytes[i + 6]) {
+ case bA:
+ // We have an AA - make sure that's the end of the field
+ if (fieldDelim != acknowledgementBytes[i + 7]) {
+ exchange.setException(new MllpInvalidAcknowledgementException(new String(acknowledgementBytes)));
+ }
+ message.setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, "AA");
+ break;
+ case bE:
+ // We have an AE
+ exchange.setException(new MllpApplicationErrorAcknowledgementException(new String(acknowledgementBytes)));
+ message.setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, "AE");
+ break;
+ case bR:
+ exchange.setException(new MllpApplicationRejectAcknowledgementException(new String(acknowledgementBytes)));
+ message.setHeader(MLLP_ACKNOWLEDGEMENT_TYPE, "AR");
+ break;
+ default:
+ exchange.setException(new MllpInvalidAcknowledgementException(new String(acknowledgementBytes)));
+ }
+ }
+
+ break;
+ }
+ }
+ }
+
+ }
+ if (-1 == msaStartIndex) {
+ // Didn't find an MSA
+ exchange.setException(new MllpInvalidAcknowledgementException(new String(acknowledgementBytes)));
+ }
+ // Check AFTER_SEND Properties
+ if (exchange.getProperty(MLLP_RESET_CONNECTION_AFTER_SEND, boolean.class)) {
+ MllpUtil.resetConnection(socket);
+ return;
+ } else if (exchange.getProperty(MLLP_CLOSE_CONNECTION_AFTER_SEND, boolean.class)) {
+ MllpUtil.closeConnection(socket);
+ }
+ }
+
+ /**
+ * Validate the TCP Connection
+ *
+ * @return null if the connection is valid, otherwise the Exception encounted checking the connection
+ */
+ Exception checkConnection() {
+ if (null == socket || socket.isClosed() || !socket.isConnected()) {
+ socket = new Socket();
+
+ try {
+ socket.setKeepAlive(endpoint.keepAlive);
+ socket.setTcpNoDelay(endpoint.tcpNoDelay);
+ if (null != endpoint.receiveBufferSize) {
+ socket.setReceiveBufferSize(endpoint.receiveBufferSize);
+ }
+ if (null != endpoint.sendBufferSize) {
+ socket.setSendBufferSize(endpoint.sendBufferSize);
+ }
+ socket.setReuseAddress(endpoint.reuseAddress);
+ socket.setSoLinger(false, -1);
+
+ // Read Timeout
+ socket.setSoTimeout(endpoint.responseTimeout);
+ } catch (SocketException e) {
+ return e;
+ }
+
+
+ SocketAddress address = new InetSocketAddress(endpoint.getHostname(), endpoint.getPort());
+ log.debug("Connecting to socket on {}", address);
+ try {
+ socket.connect(address, endpoint.connectTimeout);
+ } catch (SocketTimeoutException e) {
+ return e;
+ } catch (IOException e) {
+ return e;
+ }
+ }
+
+ return null;
+
+ }
+
+}
[2/6] camel git commit: Added camel-mllp component
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerTest.java
new file mode 100644
index 0000000..616e2d9
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpClientProducerTest.java
@@ -0,0 +1,249 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit.rule.mllp.MllpServerResource;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+public class MllpTcpClientProducerTest extends CamelTestSupport {
+ @Rule
+ public MllpServerResource mllpServer = new MllpServerResource(AvailablePortFinder.getNextAvailable());
+
+ @EndpointInject(uri = "direct://source")
+ ProducerTemplate source;
+
+ @EndpointInject(uri = "mock://acknowledged")
+ MockEndpoint acknowledged;
+
+ @EndpointInject(uri = "mock://timeout-ex")
+ MockEndpoint timeout;
+
+ @EndpointInject(uri = "mock://frame-ex")
+ MockEndpoint frame;
+
+ @Override
+ public String isMockEndpoints() {
+ return "log://netty-mllp-sender-throughput*";
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext) super.createCamelContext();
+
+ context.setUseMDCLogging(true);
+ context.setName(this.getClass().getSimpleName());
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+
+ return new RouteBuilder() {
+ int connectTimeout = 1000;
+ int responseTimeout = 1000;
+
+ @Override
+ public void configure() throws Exception {
+ errorHandler(
+ defaultErrorHandler().allowRedeliveryWhileStopping(false));
+
+ onException(MllpCorruptFrameException.class)
+ .handled(true)
+ .logHandled(false)
+ .to(frame);
+
+ onException(MllpTimeoutException.class)
+ .handled(true)
+ .logHandled(false)
+ .to(timeout);
+
+ onCompletion()
+ .onFailureOnly().log(LoggingLevel.ERROR, "Processing Failed");
+
+ from(source.getDefaultEndpoint())
+ .routeId("mllp-sender-test-route")
+ .log(LoggingLevel.INFO, "Sending Message: $simple{header[CamelHL7MessageControl]}")
+ .toF("mllp://%s:%d?connectTimeout=%d&responseTimeout=%d",
+ "0.0.0.0", mllpServer.getListenPort(), connectTimeout, responseTimeout)
+ .to(acknowledged);
+
+ from("direct://handle-timeout")
+ .log(LoggingLevel.ERROR, "Response Timeout")
+ .rollback();
+
+ }
+ };
+
+ }
+
+ @Test
+ public void testSendSingleMessage() throws Exception {
+ acknowledged.setExpectedMessageCount(1);
+ timeout.setExpectedMessageCount(0);
+ frame.setExpectedMessageCount(0);
+
+ source.sendBody(generateMessage());
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+
+ @Test
+ public void testSendMultipleMessages() throws Exception {
+ int messageCount = 5;
+ acknowledged.setExpectedMessageCount(messageCount);
+ timeout.setExpectedMessageCount(0);
+ frame.setExpectedMessageCount(0);
+
+ NotifyBuilder[] complete = new NotifyBuilder[messageCount];
+ for (int i = 0; i < messageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ for (int i = 0; i < messageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+
+ @Test
+ public void testNoResponseOnFirstMessage() throws Exception {
+ int sendMessageCount = 5;
+ acknowledged.setExpectedMessageCount(sendMessageCount - 1);
+ timeout.expectedMessageCount(1);
+ frame.setExpectedMessageCount(0);
+
+ NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
+ for (int i = 0; i < sendMessageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ mllpServer.disableResponse();
+
+ source.sendBody(generateMessage(1));
+ assertTrue("Messege 1 not completed", complete[0].matches(1, TimeUnit.SECONDS));
+
+ mllpServer.enableResponse();
+
+ for (int i = 1; i < sendMessageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testNoResponseOnNthMessage() throws Exception {
+ int sendMessageCount = 3;
+ acknowledged.setExpectedMessageCount(sendMessageCount - 1);
+ timeout.expectedMessageCount(1);
+ frame.setExpectedMessageCount(0);
+
+ NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
+ for (int i = 0; i < sendMessageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ mllpServer.disableResponse(sendMessageCount);
+
+ for (int i = 0; i < sendMessageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testMissingEndOfDataByte() throws Exception {
+ int sendMessageCount = 3;
+ acknowledged.setExpectedMessageCount(sendMessageCount - 1);
+
+ NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
+ for (int i = 0; i < sendMessageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ mllpServer.setExcludeEndOfDataModulus(sendMessageCount);
+
+ for (int i = 0; i < sendMessageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testMissingEndOfBlockByte() throws Exception {
+ int sendMessageCount = 3;
+ acknowledged.setExpectedMessageCount(sendMessageCount - 1);
+
+ NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
+ for (int i = 0; i < sendMessageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ mllpServer.setExcludeEndOfBlockModulus(sendMessageCount);
+
+ for (int i = 0; i < sendMessageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testApplicationAcceptAcknowledgement() throws Exception {
+ int sendMessageCount = 5;
+ acknowledged.setExpectedMessageCount(sendMessageCount);
+
+ NotifyBuilder[] complete = new NotifyBuilder[sendMessageCount];
+ for (int i = 0; i < sendMessageCount; ++i) {
+ complete[i] = new NotifyBuilder(context).whenDone(i + 1).create();
+ }
+
+ for (int i = 0; i < sendMessageCount; ++i) {
+ source.sendBody(generateMessage(i + 1));
+ assertTrue("Messege " + i + " not completed", complete[i].matches(1, TimeUnit.SECONDS));
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerBlueprintTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerBlueprintTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerBlueprintTest.java
new file mode 100644
index 0000000..c7dd2c9
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerBlueprintTest.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.Dictionary;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultComponentResolver;
+import org.apache.camel.spi.ComponentResolver;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.blueprint.CamelBlueprintTestSupport;
+import org.apache.camel.test.junit.rule.mllp.MllpClientResource;
+import org.apache.camel.util.KeyValueHolder;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+public class MllpTcpServerConsumerBlueprintTest extends CamelBlueprintTestSupport {
+ @Rule
+ public MllpClientResource mllpClient = new MllpClientResource();
+
+ final String receivedUri = "mock://received";
+
+ @EndpointInject(uri = receivedUri)
+ MockEndpoint received;
+
+
+ @Override
+ protected String getBlueprintDescriptor() {
+ return "OSGI-INF/blueprint/mllp-tcp-server-consumer-test.xml";
+ }
+
+ @Override
+ protected void addServicesOnStartup(Map<String, KeyValueHolder<Object, Dictionary>> services) {
+ ComponentResolver testResolver = new DefaultComponentResolver();
+
+ services.put(ComponentResolver.class.getName(), asService(testResolver, "component", "mllp"));
+ }
+
+
+ @Override
+ protected Properties useOverridePropertiesWithPropertiesComponent() {
+ mllpClient.setMllpPort(AvailablePortFinder.getNextAvailable());
+
+ Properties props = new Properties();
+
+ props.setProperty("receivedUri", receivedUri);
+ props.setProperty("mllp.port", Integer.toString(mllpClient.getMllpPort()));
+
+ return props;
+ }
+
+ /*
+ This doesn't seem to work
+ @Override
+ protected String useOverridePropertiesWithConfigAdmin(Dictionary props) throws Exception {
+ mllpClient.setMllpPort(AvailablePortFinder.getNextAvailable());
+
+ props.put("mllp.port", mllpClient.getMllpPort() );
+
+ return "MllpTcpServerConsumerBlueprintTest";
+ }
+ */
+
+ @Test
+ public void testReceiveMultipleMessages() throws Exception {
+ int sendMessageCount = 5;
+ received.expectedMinimumMessageCount(5);
+
+ mllpClient.connect();
+
+ for (int i = 1; i <= sendMessageCount; ++i) {
+ mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(i));
+ }
+
+ mllpClient.close();
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerConnectionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerConnectionTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerConnectionTest.java
new file mode 100644
index 0000000..fab33a5
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerConnectionTest.java
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class MllpTcpServerConsumerConnectionTest extends CamelTestSupport {
+ int mllpPort;
+ @EndpointInject(uri = "mock://result")
+ MockEndpoint result;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ mllpPort = AvailablePortFinder.getNextAvailable();
+
+ return new RouteBuilder() {
+ String routeId = "mllp-receiver";
+
+ String host = "0.0.0.0";
+ int port = mllpPort;
+
+ public void configure() {
+ fromF("mllp:%d?autoAck=false", port)
+ .log(LoggingLevel.INFO, routeId, "Receiving: ${body}")
+ .to(result);
+ }
+ };
+
+ }
+
+ /**
+ * Simulate a Load Balancer Probe
+ * <p/>
+ * Load Balancers check the status of a port by establishing and closing a TCP connection periodically. The time
+ * between these probes can normally be configured, but it is typically set to about 15-sec. Since there could be
+ * a large number of port that are being probed, the logging from the connect/disconnect operations can drown-out
+ * more useful information.
+ * <p/>
+ * Watch the logs when running this test to verify that the log output will be acceptable when a load balancer
+ * is probing the port.
+ * <p/>
+ * TODO: Need to add a custom Log4j Appender that can verify the logging is acceptable
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testConnectWithoutData() throws Exception {
+ result.setExpectedCount(0);
+ int connectionCount = 10;
+
+ Socket dummyLoadBalancerSocket = null;
+ SocketAddress address = new InetSocketAddress("localhost", mllpPort);
+ int connectTimeout = 5000;
+ try {
+ for (int i = 1; i <= connectionCount; ++i) {
+ log.debug("Creating connection #{}", i);
+ dummyLoadBalancerSocket = new Socket();
+ dummyLoadBalancerSocket.connect(address, connectTimeout);
+ log.debug("Closing connection #{}", i);
+ dummyLoadBalancerSocket.close();
+ Thread.sleep(1000);
+ }
+ } finally {
+ if (null != dummyLoadBalancerSocket) {
+ try {
+ dummyLoadBalancerSocket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing dummy load balancer socket", ex);
+ }
+ }
+ }
+
+ assertMockEndpointsSatisfied(15, TimeUnit.SECONDS);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerMulitpleTcpPacketTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerMulitpleTcpPacketTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerMulitpleTcpPacketTest.java
new file mode 100644
index 0000000..afb3080
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerMulitpleTcpPacketTest.java
@@ -0,0 +1,120 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit.rule.mllp.MllpClientResource;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.apache.camel.test.mllp.PassthroughProcessor;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+public class MllpTcpServerConsumerMulitpleTcpPacketTest extends CamelTestSupport {
+ @Rule
+ public MllpClientResource mllpClient = new MllpClientResource();
+
+ @EndpointInject(uri = "mock://result")
+ MockEndpoint result;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext) super.createCamelContext();
+
+ context.setUseMDCLogging(true);
+ context.setName(this.getClass().getSimpleName());
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ final int groupInterval = 1000;
+ final boolean groupActiveOnly = false;
+
+ mllpClient.setMllpHost("localhost");
+ mllpClient.setMllpPort(AvailablePortFinder.getNextAvailable());
+
+
+ return new RouteBuilder() {
+ String routeId = "mllp-receiver";
+
+ @Override
+ public void configure() throws Exception {
+
+ onCompletion()
+ .log(LoggingLevel.DEBUG, routeId, "Test route complete");
+
+ fromF("mllp://%s:%d",
+ mllpClient.getMllpHost(), mllpClient.getMllpPort())
+ .routeId(routeId)
+ .process(new PassthroughProcessor("Before send to result"))
+ .to(result)
+ .toF("log://%s?level=INFO&groupInterval=%d&groupActiveOnly=%b", routeId, groupInterval, groupActiveOnly)
+ .log(LoggingLevel.DEBUG, routeId, "Test route received message");
+
+ }
+ };
+ }
+
+ @Test
+ public void testReceiveSingleMessage() throws Exception {
+ mllpClient.connect();
+
+ String message = generateMessage();
+ result.expectedBodiesReceived(message);
+
+ mllpClient.sendFramedDataInMultiplePackets(message, (byte) '\r');
+ String acknowledgement = mllpClient.receiveFramedData();
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+
+ Assert.assertThat("Should be acknowledgment for message 1", acknowledgement, CoreMatchers.containsString(String.format("MSA|AA|00001")));
+ }
+
+
+ @Test
+ public void testReceiveMultipleMessages() throws Exception {
+ int sendMessageCount = 1000;
+ result.expectedMinimumMessageCount(5);
+
+ mllpClient.connect();
+
+ for (int i = 1; i <= sendMessageCount; ++i) {
+ String testMessage = generateMessage(i);
+ result.message(i - 1).body().isEqualTo(testMessage);
+ mllpClient.sendFramedDataInMultiplePackets(testMessage, (byte) '\r');
+ String acknowledgement = mllpClient.receiveFramedData();
+ Assert.assertThat("Should be acknowledgment for message " + i, acknowledgement, CoreMatchers.containsString(String.format("MSA|AA|%05d", i)));
+ }
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerTest.java
new file mode 100644
index 0000000..2ddd25c
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerConsumerTest.java
@@ -0,0 +1,169 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit.rule.mllp.MllpClientResource;
+import org.apache.camel.test.junit.rule.mllp.MllpJUnitResourceException;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static org.apache.camel.test.mllp.Hl7MessageGenerator.generateMessage;
+
+public class MllpTcpServerConsumerTest extends CamelTestSupport {
+ @Rule
+ public MllpClientResource mllpClient = new MllpClientResource();
+
+ @EndpointInject(uri = "mock://result")
+ MockEndpoint result;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ DefaultCamelContext context = (DefaultCamelContext) super.createCamelContext();
+
+ context.setUseMDCLogging(true);
+ context.setName(this.getClass().getSimpleName());
+
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+
+ mllpClient.setMllpHost("localhost");
+ mllpClient.setMllpPort(AvailablePortFinder.getNextAvailable());
+
+ return new RouteBuilder() {
+ int connectTimeout = 500;
+ int responseTimeout = 5000;
+
+ @Override
+ public void configure() throws Exception {
+ String routeId = "mllp-test-receiver-route";
+
+ onCompletion()
+ .toF("log:%s?level=INFO&showAll=true", routeId)
+ .log(LoggingLevel.INFO, routeId, "Test route complete");
+
+ fromF("mllp://%s:%d?autoAck=true&connectTimeout=%d&responseTimeout=%d",
+ mllpClient.getMllpHost(), mllpClient.getMllpPort(), connectTimeout, responseTimeout)
+ .routeId(routeId)
+ .log(LoggingLevel.INFO, routeId, "Test route received message")
+ .to(result);
+
+ }
+ };
+ }
+
+ @Test
+ public void testReceiveSingleMessage() throws Exception {
+ result.expectedMessageCount(1);
+
+ mllpClient.connect();
+
+ mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(), 10000);
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testReceiveSingleMessageWithDelayAfterConnection() throws Exception {
+ result.expectedMinimumMessageCount(1);
+
+ mllpClient.connect();
+
+ Thread.sleep(5000);
+ mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(), 10000);
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testReceiveMultipleMessages() throws Exception {
+ int sendMessageCount = 5;
+ result.expectedMinimumMessageCount(5);
+
+ mllpClient.connect();
+
+ for (int i = 1; i <= sendMessageCount; ++i) {
+ mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(i));
+ }
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testOpenMllpEnvelopeWithReset() throws Exception {
+ result.expectedMessageCount(4);
+ NotifyBuilder notify1 = new NotifyBuilder(context).whenDone(2).create();
+ NotifyBuilder notify2 = new NotifyBuilder(context).whenDone(5).create();
+
+ mllpClient.connect();
+ mllpClient.setSoTimeout(10000);
+
+ log.info("Sending TEST_MESSAGE_1");
+ String acknowledgement1 = mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(1));
+
+ log.info("Sending TEST_MESSAGE_2");
+ String acknowledgement2 = mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(2));
+
+ assertTrue("First two normal exchanges did not complete", notify1.matches(10, TimeUnit.SECONDS));
+
+ log.info("Sending TEST_MESSAGE_3");
+ mllpClient.setSendEndOfBlock(false);
+ mllpClient.setSendEndOfData(false);
+ // Acknowledgement won't come here
+ try {
+ mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(3));
+ } catch (MllpJUnitResourceException resourceEx) {
+ log.info("Expected exception reading response");
+ }
+ mllpClient.disconnect();
+ Thread.sleep(1000);
+ mllpClient.connect();
+
+ log.info("Sending TEST_MESSAGE_4");
+ mllpClient.setSendEndOfBlock(true);
+ mllpClient.setSendEndOfData(true);
+ String acknowledgement4 = mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(4));
+
+ log.info("Sending TEST_MESSAGE_5");
+ String acknowledgement5 = mllpClient.sendMessageAndWaitForAcknowledgement(generateMessage(5));
+
+ assertTrue("Remaining exchanges did not complete", notify2.matches(10, TimeUnit.SECONDS));
+
+ assertMockEndpointsSatisfied(10, TimeUnit.SECONDS);
+
+ assertTrue("Should be acknowledgment for message 1", acknowledgement1.contains("MSA|AA|00001"));
+ assertTrue("Should be acknowledgment for message 2", acknowledgement2.contains("MSA|AA|00002"));
+ // assertTrue("Should be acknowledgment for message 3", acknowledgement3.contains("MSA|AA|00003"));
+ assertTrue("Should be acknowledgment for message 4", acknowledgement4.contains("MSA|AA|00004"));
+ assertTrue("Should be acknowledgment for message 5", acknowledgement5.contains("MSA|AA|00005"));
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerProducerBlueprintTest.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerProducerBlueprintTest.java b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerProducerBlueprintTest.java
new file mode 100644
index 0000000..dc796a5
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/component/mllp/MllpTcpServerProducerBlueprintTest.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mllp;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.blueprint.CamelBlueprintTestSupport;
+import org.junit.Ignore;
+
+@Ignore(value = "Not Yet Implemented")
+// TODO: Implement this
+public class MllpTcpServerProducerBlueprintTest extends CamelBlueprintTestSupport {
+ @EndpointInject(uri = "mock://target")
+ MockEndpoint target;
+
+ @Override
+ protected String getBlueprintDescriptor() {
+ return "OSGI-INF/blueprint/mllp-tcp-server-producer.xml";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
new file mode 100644
index 0000000..8157e69
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpClientResource.java
@@ -0,0 +1,461 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.junit.rule.mllp;
+
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+
+import org.junit.rules.ExternalResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MLLP Test Client packaged as a JUnit Rule
+ *
+ * The client can be configured to simulate a large number
+ * of error conditions.
+ */
+public class MllpClientResource extends ExternalResource {
+
+ static final char START_OF_BLOCK = 0x0b;
+ static final char END_OF_BLOCK = 0x1c;
+ static final char END_OF_DATA = 0x0d;
+ static final int END_OF_STREAM = -1;
+
+ Logger log = LoggerFactory.getLogger(this.getClass());
+
+ Socket clientSocket;
+ InputStream inputStream;
+ OutputStream outputStream;
+
+ String mllpHost = "0.0.0.0";
+ int mllpPort = -1;
+
+ boolean sendStartOfBlock = true;
+ boolean sendEndOfBlock = true;
+ boolean sendEndOfData = true;
+
+ int connectTimeout = 5000;
+ int soTimeout = 5000;
+ boolean reuseAddress;
+ boolean tcpNoDelay = true;
+
+
+ /**
+ * Use this constructor to avoid having the connection started by JUnit (since the port is still -1)
+ */
+ public MllpClientResource() {
+
+ }
+
+ public MllpClientResource(int port) {
+ this.mllpPort = port;
+ }
+
+ public MllpClientResource(String host, int port) {
+ this.mllpHost = host;
+ this.mllpPort = port;
+ }
+
+ @Override
+ protected void before() throws Throwable {
+ if (0 < mllpPort) {
+ this.connect();
+ }
+
+ super.before();
+ }
+
+ @Override
+ protected void after() {
+ super.after();
+ this.disconnect();
+ }
+
+ public void connect() {
+ this.connect(this.connectTimeout);
+ }
+
+ public void connect(int connectTimeout) {
+ try {
+ clientSocket = new Socket();
+
+ clientSocket.connect(new InetSocketAddress(mllpHost, mllpPort), connectTimeout);
+
+ clientSocket.setSoTimeout(soTimeout);
+ clientSocket.setSoLinger(false, -1);
+ clientSocket.setReuseAddress(reuseAddress);
+ clientSocket.setTcpNoDelay(tcpNoDelay);
+
+ inputStream = clientSocket.getInputStream();
+ outputStream = new BufferedOutputStream(clientSocket.getOutputStream(), 2048);
+ } catch (IOException e) {
+ String errorMessage = String.format("Unable to establish connection to %s:%s", mllpHost, mllpPort);
+ log.error(errorMessage, e);
+ throw new MllpJUnitResourceException(errorMessage, e);
+ }
+ }
+
+ public void close() {
+ this.disconnect();
+ return;
+ }
+
+ public void reset() {
+ try {
+ clientSocket.setSoLinger(true, 0);
+ } catch (SocketException socketEx) {
+ log.warn("Exception encountered setting set SO_LINGER to force a TCP reset", socketEx);
+ }
+ this.disconnect();
+ return;
+ }
+
+ public void disconnect() {
+ try {
+ if (null != inputStream) {
+ clientSocket.close();
+ }
+ } catch (IOException e) {
+ log.warn(String.format("Exception encountered closing connection to {}:{}", mllpHost, mllpPort), e);
+ } finally {
+ inputStream = null;
+ outputStream = null;
+ clientSocket = null;
+ }
+ }
+
+ public boolean isConnected() {
+ return clientSocket.isConnected() && !clientSocket.isClosed();
+ }
+
+ public void sendData(String data) {
+ boolean disconnectAfterSend = false;
+ this.sendData(data, disconnectAfterSend);
+ }
+
+ public void sendData(String data, boolean disconnectAfterSend) {
+ byte[] payloadBytes = data.getBytes();
+
+ try {
+ outputStream.write(payloadBytes, 0, payloadBytes.length);
+ } catch (IOException e) {
+ log.error("Unable to send raw string", e);
+ throw new MllpJUnitResourceException("Unable to send raw string", e);
+ }
+
+ if (disconnectAfterSend) {
+ log.warn("Closing TCP connection");
+ disconnect();
+ }
+ }
+
+ public void sendFramedData(String hl7Message) {
+ boolean disconnectAfterSend = false;
+ this.sendFramedData(hl7Message, disconnectAfterSend);
+ }
+
+ public void sendFramedData(String hl7Message, boolean disconnectAfterSend) {
+ if (null == clientSocket) {
+ this.connect();
+ }
+
+ if (!clientSocket.isConnected()) {
+ throw new MllpJUnitResourceException("Cannot send message - client is not connected");
+ }
+ if (null == outputStream) {
+ throw new MllpJUnitResourceException("Cannot send message - output stream is null");
+ }
+ byte[] payloadBytes = hl7Message.getBytes();
+ try {
+ if (sendStartOfBlock) {
+ outputStream.write(START_OF_BLOCK);
+ } else {
+ log.warn("Not sending START_OF_BLOCK");
+ }
+ outputStream.write(payloadBytes, 0, payloadBytes.length);
+ if (sendEndOfBlock) {
+ outputStream.write(END_OF_BLOCK);
+ } else {
+ log.warn("Not sending END_OF_BLOCK");
+ }
+ if (sendEndOfData) {
+ outputStream.write(END_OF_DATA);
+ } else {
+ log.warn("Not sending END_OF_DATA");
+ }
+ outputStream.flush();
+ } catch (IOException e) {
+ log.error("Unable to send HL7 message", e);
+ throw new MllpJUnitResourceException("Unable to send HL7 message", e);
+ }
+
+ if (disconnectAfterSend) {
+ log.warn("Closing TCP connection");
+ disconnect();
+ }
+ }
+
+ public void sendFramedDataInMultiplePackets(String hl7Message, byte flushByte) {
+ sendFramedDataInMultiplePackets(hl7Message, flushByte, false);
+ }
+
+ public void sendFramedDataInMultiplePackets(String hl7Message, byte flushByte, boolean disconnectAfterSend) {
+ if (null == clientSocket) {
+ this.connect();
+ }
+
+ if (!clientSocket.isConnected()) {
+ throw new MllpJUnitResourceException("Cannot send message - client is not connected");
+ }
+ if (null == outputStream) {
+ throw new MllpJUnitResourceException("Cannot send message - output stream is null");
+ }
+ byte[] payloadBytes = hl7Message.getBytes();
+ try {
+ if (sendStartOfBlock) {
+ outputStream.write(START_OF_BLOCK);
+ } else {
+ log.warn("Not sending START_OF_BLOCK");
+ }
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ outputStream.write(payloadBytes[i]);
+ if (flushByte == payloadBytes[i]) {
+ outputStream.flush();
+ }
+ }
+ if (sendEndOfBlock) {
+ outputStream.write(END_OF_BLOCK);
+ } else {
+ log.warn("Not sending END_OF_BLOCK");
+ }
+ if (sendEndOfData) {
+ outputStream.write(END_OF_DATA);
+ } else {
+ log.warn("Not sending END_OF_DATA");
+ }
+ outputStream.flush();
+ } catch (IOException e) {
+ log.error("Unable to send HL7 message", e);
+ throw new MllpJUnitResourceException("Unable to send HL7 message", e);
+ }
+
+ if (disconnectAfterSend) {
+ log.warn("Closing TCP connection");
+ disconnect();
+ }
+ }
+
+ public String receiveFramedData() throws SocketException, SocketTimeoutException {
+ return receiveFramedData(soTimeout);
+ }
+
+ public String receiveFramedData(int timout) throws SocketException, SocketTimeoutException {
+ if (!isConnected()) {
+ throw new MllpJUnitResourceException("Cannot receive acknowledgement - client is not connected");
+ }
+ if (null == outputStream) {
+ throw new MllpJUnitResourceException("Cannot receive acknowledgement - output stream is null");
+ }
+
+ clientSocket.setSoTimeout(timout);
+ StringBuilder acknowledgement = new StringBuilder();
+ try {
+ int firstByte = inputStream.read();
+ if (START_OF_BLOCK != firstByte) {
+ if (isConnected()) {
+ if (END_OF_STREAM == firstByte) {
+ log.warn("END_OF_STREAM reached while waiting for START_OF_BLOCK - closing socket");
+ try {
+ clientSocket.close();
+ } catch (Exception ex) {
+ log.warn("Exception encountered closing socket after receiving END_OF_STREAM while waiting for START_OF_BLOCK");
+ }
+ return "";
+ } else {
+ log.error("Acknowledgement did not start with START_OF_BLOCK: {}", firstByte);
+ throw new MllpJUnitResourceCorruptFrameException("Message did not start with START_OF_BLOCK");
+ }
+ } else {
+ throw new MllpJUnitResourceException("Connection terminated");
+ }
+ }
+ boolean readingMessage = true;
+ while (readingMessage) {
+ int nextByte = inputStream.read();
+ switch (nextByte) {
+ case -1:
+ throw new MllpJUnitResourceCorruptFrameException("Reached end of stream before END_OF_BLOCK");
+ case START_OF_BLOCK:
+ throw new MllpJUnitResourceCorruptFrameException("Received START_OF_BLOCK before END_OF_BLOCK");
+ case END_OF_BLOCK:
+ if (END_OF_DATA != inputStream.read()) {
+ throw new MllpJUnitResourceCorruptFrameException("END_OF_BLOCK was not followed by END_OF_DATA");
+ }
+ readingMessage = false;
+ break;
+ default:
+ acknowledgement.append((char) nextByte);
+ }
+ }
+ } catch (SocketTimeoutException timeoutEx) {
+ if (0 < acknowledgement.length()) {
+ log.error("Timeout waiting for acknowledgement", timeoutEx);
+ } else {
+ log.error("Timeout while reading acknowledgement\n" + acknowledgement.toString().replace('\r', '\n'), timeoutEx);
+ }
+ throw new MllpJUnitResourceTimeoutException("Timeout while reading acknowledgement", timeoutEx);
+ } catch (IOException e) {
+ log.error("Unable to read HL7 acknowledgement", e);
+ throw new MllpJUnitResourceException("Unable to read HL7 acknowledgement", e);
+ }
+
+ return acknowledgement.toString();
+ }
+
+ public String receiveData() throws SocketException, SocketTimeoutException {
+ return receiveData(soTimeout);
+ }
+
+ public String receiveData(int timeout) throws SocketException, SocketTimeoutException {
+ clientSocket.setSoTimeout(timeout);
+ StringBuilder availableInput = new StringBuilder();
+
+ try {
+ do {
+ availableInput.append((char) inputStream.read());
+ } while (0 < inputStream.available());
+ } catch (SocketTimeoutException timeoutEx) {
+ log.error("Timeout while receiving available input", timeoutEx);
+ throw new MllpJUnitResourceTimeoutException("Timeout while receiving available input", timeoutEx);
+ } catch (IOException e) {
+ log.warn("Exception encountered eating available input", e);
+ throw new MllpJUnitResourceException("Exception encountered eating available input", e);
+ }
+
+ return availableInput.toString();
+ }
+
+ public String eatData() throws SocketException, SocketTimeoutException {
+ return eatData(soTimeout);
+ }
+
+ public String eatData(int timeout) throws SocketException {
+ clientSocket.setSoTimeout(timeout);
+
+ StringBuilder availableInput = new StringBuilder();
+ try {
+ while (0 < inputStream.available()) {
+ availableInput.append((char) inputStream.read());
+ }
+ } catch (IOException e) {
+ log.warn("Exception encountered eating available input", e);
+ throw new MllpJUnitResourceException("Exception encountered eating available input", e);
+ }
+
+ return availableInput.toString();
+ }
+
+ public String sendMessageAndWaitForAcknowledgement(String hl7Data) throws SocketException, SocketTimeoutException {
+ sendFramedData(hl7Data);
+ return receiveFramedData();
+ }
+
+ public String sendMessageAndWaitForAcknowledgement(String hl7Data, int acknwoledgementTimeout) throws SocketException, SocketTimeoutException {
+ sendFramedData(hl7Data);
+ return receiveFramedData(acknwoledgementTimeout);
+ }
+
+ public String getMllpHost() {
+ return mllpHost;
+ }
+
+ public void setMllpHost(String mllpHost) {
+ this.mllpHost = mllpHost;
+ }
+
+ public int getMllpPort() {
+ return mllpPort;
+ }
+
+ public void setMllpPort(int mllpPort) {
+ this.mllpPort = mllpPort;
+ }
+
+ public boolean isSendStartOfBlock() {
+ return sendStartOfBlock;
+ }
+
+ public void setSendStartOfBlock(boolean sendStartOfBlock) {
+ this.sendStartOfBlock = sendStartOfBlock;
+ }
+
+ public boolean isSendEndOfBlock() {
+ return sendEndOfBlock;
+ }
+
+ public void setSendEndOfBlock(boolean sendEndOfBlock) {
+ this.sendEndOfBlock = sendEndOfBlock;
+ }
+
+ public boolean isSendEndOfData() {
+ return sendEndOfData;
+ }
+
+ public void setSendEndOfData(boolean sendEndOfData) {
+ this.sendEndOfData = sendEndOfData;
+ }
+
+ public int getConnectTimeout() {
+ return connectTimeout;
+ }
+
+ public void setConnectTimeout(int connectTimeout) {
+ this.connectTimeout = connectTimeout;
+ }
+
+ public int getSoTimeout() {
+ return soTimeout;
+ }
+
+ public void setSoTimeout(int soTimeout) {
+ this.soTimeout = soTimeout;
+ }
+
+ public boolean isReuseAddress() {
+ return reuseAddress;
+ }
+
+ public void setReuseAddress(boolean reuseAddress) {
+ this.reuseAddress = reuseAddress;
+ }
+
+ public boolean isTcpNoDelay() {
+ return tcpNoDelay;
+ }
+
+ public void setTcpNoDelay(boolean tcpNoDelay) {
+ this.tcpNoDelay = tcpNoDelay;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceCorruptFrameException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceCorruptFrameException.java b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceCorruptFrameException.java
new file mode 100644
index 0000000..9fbbdfe
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceCorruptFrameException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.junit.rule.mllp;
+
+/**
+ * Thrown when one of the MLLP JUnit Rules encounters a corrupt MLLP Frame
+ */
+public class MllpJUnitResourceCorruptFrameException extends MllpJUnitResourceException {
+ public MllpJUnitResourceCorruptFrameException(String message) {
+ super(message);
+ }
+
+ public MllpJUnitResourceCorruptFrameException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpJUnitResourceCorruptFrameException(Throwable cause) {
+ super(cause);
+ }
+
+ public MllpJUnitResourceCorruptFrameException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceException.java b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceException.java
new file mode 100644
index 0000000..e5a6c4e
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.junit.rule.mllp;
+
+/**
+ * Base Exception for MLLP JUnit Rules
+ */
+public class MllpJUnitResourceException extends RuntimeException {
+ public MllpJUnitResourceException(String message) {
+ super(message);
+ }
+
+ public MllpJUnitResourceException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpJUnitResourceException(Throwable cause) {
+ super(cause);
+ }
+
+ public MllpJUnitResourceException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bd1661b2/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceTimeoutException.java
----------------------------------------------------------------------
diff --git a/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceTimeoutException.java b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceTimeoutException.java
new file mode 100644
index 0000000..8e6a2f6
--- /dev/null
+++ b/components/camel-mllp/src/test/java/org/apache/camel/test/junit/rule/mllp/MllpJUnitResourceTimeoutException.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.junit.rule.mllp;
+
+/**
+ * Thrown when one of the MLLP JUnit Rules encounters an unexpected timeout
+ */
+public class MllpJUnitResourceTimeoutException extends MllpJUnitResourceException {
+ public MllpJUnitResourceTimeoutException(String message) {
+ super(message);
+ }
+
+ public MllpJUnitResourceTimeoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public MllpJUnitResourceTimeoutException(Throwable cause) {
+ super(cause);
+ }
+
+ public MllpJUnitResourceTimeoutException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ }
+}