Bidirectional Tcp connector

Rassul Hessampour

2012-06-20 12:12:32 UTC

Dear david.
 
Excuse me for being late in response.
 
Here is settings in config files :
 
1) Connector definition:
 
<code>
 
<tcp:connector name="myConnector" sendTcpNoDelay="true"
 
sendBufferSize="8192" receiveBufferSize="8192" createMultipleTransactedReceivers="true"
 
numberOfConcurrentTransactedReceivers="30" keepSendSocketOpen="true">
 
<spring:property name="tcpProtocol" ref="myProtocol"/>
 
</tcp:connector>
 
</code>
 
2) Dispatcher endpoint :
 
<code>
 
<tcp:endpoint name="dispatchEndpoint" encoding="8859_1"
 
address="tcp://127.0.0.1:5002" connector-ref="myConnector" remoteSync="true"
 
remoteSyncTimeout="6000">
 
</tcp:endpoint>
 
</code>
 
3) Dispatcher class :
 
<code>
 
public class MyTcpMessageDispatcher extends AbstractMessageDispatcher {
 
private Socket socket;
 
private final TcpConnector connector;
 
public MyTcpMessageDispatcher(OutboundEndpoint endpoint) {
 
super(endpoint);
 
this.connector = (TcpConnector) endpoint.getConnector();
 
SocketListener listener = new SocketListener();
 
new Thread(listener).start();
 
}
 
protected void doConnect() throws Exception {
 
disposing.set(false);
 
}
 
protected void doDisconnect() throws ConnectException {
 
disposing.set(true);
 
try {
 
if (socket != null) {
 
if (logger.isDebugEnabled()) {
 
logger.debug("Closing: " + socket);
 
}
 
socket.close();
 
}
 
} catch (IOException e) {
 
logger.warn("Failed to close server socket: " + e.getMessage(), e);
 
}
 
}
 
protected synchronized MuleMessage doSend(MuleEvent event) throws Exception {
 
if(socket == null || !socket.isConnected()) {
 
socket = getSocket(event.getEndpoint());
 
}
 
dispatchToSocket(socket, event);
 
if (useRemoteSync(event)) {
 
return receiveAndProcessSyncResponseFromSocket(event.getTimeout(), event.getMessage(),
 
event.getEndpoint().getEndpointURI());
 
}
 
else {
 
return event.getMessage();
 
}
 
}
 
private MuleMessage receiveAndProcessSyncResponseFromSocket(int timeout, MuleMessage muleMessage,
 
EndpointURI endpointURI) throws Exception {
 
try {
 
Object result = receiveFromSocket(socket, timeout, endpoint);
 
if (!isResponseMessage(result)) {
 
byte[] response = createServerResponseMessage((byte[]) result);
 
write(socket, response);
 
return null;
 
}
 
else {
 
if (result == null) {
 
return null;
 
}
 
if (result instanceof MuleMessage) {
 
return (MuleMessage) result;
 
}
 
return MuleMessageUtil.createDefaultMessageFromMessage(muleMessage,
 
connector.getMessageAdapter(result));
 
}
 
}
 
catch (SocketTimeoutException e) {
 
logger.info("Socket timed out normally while doing a synchronous receive on endpointUri: "
 
+ endpointURI);
 
throw new ResponseTimeoutException("Remote system timeout happened!");
 
}
 
catch (NullPointerException ex) {
 
logger.info("Nothing received from endpointUri: "
 
+ endpointURI);
 
return null;
 
}
 
catch (Exception ex) {
 
logger.info("Error receiving synch response from endpointUri: "
 
+ endpointURI);
 
throw new EndpointException(ex);
 
}
 
}
 
private boolean isResponseMessage(Object receivedMessage) {
 
//check if received data is a response sent by remote server or is a request received from server;
 
return true;
 
}
 
protected synchronized void doDispatch(MuleEvent event) throws Exception {
 
if(socket == null || !socket.isConnected()) {
 
socket = getSocket(event.getEndpoint());
 
}
 
try {
 
dispatchToSocket(socket, event);
 
}
 
finally {
 
}
 
}
 
private void dispatchToSocket(Socket socket, MuleEvent event) throws Exception {
 
Object payload = event.transformMessage();
 
write(socket, payload);
 
}
 
private void write(Socket socket, Object data) throws IOException, TransformerException {
 
BufferedOutputStream bos = new BufferedOutputStream(socket.getOutputStream());
 
connector.getTcpProtocol().write(bos, data);
 
bos.flush();
 
}
 
protected static Object receiveFromSocket(final Socket socket, int timeout, final ImmutableEndpoint endpoint)
 
throws IOException {
 
final TcpConnector connector = (TcpConnector) endpoint.getConnector();
 
DataInputStream underlyingIs = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
 
if (timeout >= 0) {
 
socket.setSoTimeout(timeout);
 
}
 
try {
 
return connector.getTcpProtocol().read(underlyingIs);
 
}
 
finally {
 
}
 
}
 
protected void doDispose() {
 
try {
 
if (socket != null && !socket.isClosed()) {
 
if (logger.isDebugEnabled()) {
 
logger.debug("Closing: " + socket);
 
}
 
socket.close();
 
}
 
socket = null;
 
} catch (Exception e) {
 
logger.error(new DisposeException(TcpMessages.failedToCloseSocket(), e, this));
 
}
 
logger.info("Closed Tcp port");
 
}
 
protected Socket getSocket(ImmutableEndpoint endpoint) throws IOException {
 
return new Socket(endpoint.getEndpointURI().getHost(),
 
endpoint.getEndpointURI().getPort());
 
}
 
private byte[] createServerResponseMessage(byte[] requestMessage) throws Exception {
 
// create a response appropriate for remote server requests
 
return new byte[20];
 
}
 
protected class SocketListener implements Runnable {
 
public void run() {
 
while (true) {
 
try {
 
Thread.sleep(30000);
 
receiveAndProcessSyncResponseFromSocket(endpoint.getRemoteSyncTimeout(), null, endpoint.getEndpointURI());
 
}
 
catch (InterruptedException e) {
 
e.printStackTrace();
 
}
 
catch (Exception e) {
 
e.printStackTrace();
 
}
 
}
 
}
 
}
 
}
 
</code>
 
I'm still waiting for your recommendations.
 
Sincerely.

---------------------------------------------------------------------
To unsubscribe from this list, please visit:

http://xircles.codehaus.org/manage_email