Forget it, here is the code. So sorry for so many emails.
package org.apache.servicemix.components.tcp;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.SocketChannel;
import javax.jbi.JBIException;
import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.xml.transform.TransformerException;
import org.apache.servicemix.MessageExchangeListener;
import org.apache.servicemix.components.util.ComponentSupport;
import org.springframework.beans.factory.InitializingBean;
public abstract class TcpComponentSupport extends ComponentSupport
implements
MessageExchangeListener, InitializingBean {
/**
* Logger for this class
*/
private static final Logger log = Logger
.getLogger(TcpComponentSupport.class);
private String host;
private int port = -1;
private boolean useNIO = false;
private int retryInterval = 10000;
protected SocketChannel socketChannel = null;
private Socket socket = null;
protected InputStream in = null;
protected OutputStream out = null;
protected boolean running = false;
private TcpMarshaler marshaler = new TcpMarshaler();
protected abstract void processConnection() throws JBIException,
IOException;
protected abstract void processIncomingMessage(String tcpMessage);
public void setHost(String host) {
this.host = host;
} // en dof setHost(String)(
public void setPort(int port) {
this.port = port;
} // en dof setPort(int)
public void setUseNIO(boolean useNIO) {
this.useNIO = useNIO;
} // end of setUserNIO(boolean)
public void setRetryInterval(int retryInterval) {
this.retryInterval = retryInterval;
} // end of setRetryInterval(int)
protected boolean getUseNIO() {
return this.useNIO;
} // end of getUseNIO()
protected String getHost() {
return this.host;
} // en dof getHost()
protected int getPort() {
return this.port;
} // en dof getPort()
public void onMessageExchange(MessageExchange exchange)
throws MessagingException {
String tcpMessage = null;
try {
tcpMessage = this.marshaler.fromNMS(exchange.getMessage("in"));
this.processIncomingMessage(tcpMessage);
this.done(exchange);
} catch (TransformerException e) {
throw new MessagingException(e);
} // end of try /catch
} // end of onMessageExchange(MessageExchange)
public void afterPropertiesSet() throws Exception {
if (host == null) {
throw new IllegalArgumentException("You must specify the host
property.");
} // end of if
if (this.port < 0) {
throw new IllegalArgumentException("You must specify the port
property.");
} // end of if
} // end of afterPropertiesSet()
public void start() throws JBIException {
if (this.useNIO) {
// configure the NIO SocketChannel
this.startNIO();
} else {
// this is the default
// we will simply open a basic socket and create the Input and
Ooutput Streams
this.startNonNIO();
} // end of if / else
} // end of start()
public void stop() throws JBIException {
if (useNIO) {
this.stopNIO();
} else {
this.stopNonNIO();
} // ne do if / else
} // end of stop()
protected void sendToBus(String message) {
log.debug("Inside sendToBus(), sending message: " + message);
try {
InOnly exchange =
this.getExchangeFactory().createInOnlyExchange();
NormalizedMessage normalizedMessage = exchange.createMessage();
exchange.setInMessage(normalizedMessage);
this.marshaler.toNMS(normalizedMessage, message);
this.getDeliveryChannel().sendSync(exchange);
//done(exchange);
} catch (MessagingException me) {
throw new TcpConnectionException(me, message);
} // end of try / catch
} // end of sendToBus(String)
/**
* This is an empty implementation of the init().
* This method exists so that the concrete class can override it
* if there is something that needs to be done to initialize the
communication
* with the server.
*
* This method is called after the connection has been made and after
the
* Input and Output Streams have been assigned.
*
* This method is called right before the processConnection().
*/
protected void init() {
// nothing is done here
// if something needs to be done here in the concrete class
// this method is called after the connection is made and the Input
and Output Streams
// have been set but before the processConnection()
} // end of init()
/**
* This is an empty implementation of the cleanup().
*
* This method exists so the the concrete class can override it
* if there is something that needs to be done to cleanup resources
before the
* connection is shutdown.
*
* This method is called while the connection is still active and the
Input
* and Output Streams are still available.
*/
protected void cleanup() {
// nothing is done here
// if something needs to be done here in the concrete class
} // end of init()
private void startNIO() throws JBIException {
if (log.isInfoEnabled()) {
log.info("Using NIO to connect to " + this.host + ":" +
this.port);
} // en dof if
if (!this.running) {
this.connectNIO();
try {
this.init();
this.processConnection();
} catch (SocketException e) {
this.running = false;
this.startNIO();
} catch (IOException ioe) {
throw new JBIException(ioe);
} // end of try / catch
} // end of if
} // end of startNIO()
private void connectNIO() throws JBIException {
try {
this.socketChannel = SocketChannel.open();
this.socketChannel.configureBlocking(false);
this.socketChannel.connect(new InetSocketAddress(this.host,
this.port));
int count = 0;
while (!this.socketChannel.finishConnect()) {
log.debug("Finishing connection: " + (count++) + " times");
} // end of while loop
this.running = true;
} catch (ConnectException ce) {
log.warn(this.host + ":" + this.port + " unavailable, waiting "
+ this.retryInterval + " milliseconds before trying again.");
this.running = false;
try {
Thread.sleep(this.retryInterval);
} catch (InterruptedException e) {}
this.connectNIO();
} catch (Exception e) {
log.error("Exception while trying to make a tcp connection to "
+ this.host + ":" + this.port);
log.error(e);
this.running = false;
throw new JBIException("Failed to connect to " + this.host + ":"
+ this.port, e);
} // end of try / catch
} // end of connectNIO()
private void startNonNIO() throws JBIException {
if (log.isInfoEnabled()) {
log.info("Using basic sockets to connect to " + this.host + ":"
+ this.port);
} // en dof if
if (!this.running) {
this.connectNonNIO();
try {
this.init();
this.processConnection();
} catch (SocketException e) {
this.running = false;
this.startNonNIO();
} catch (IOException ioe) {
throw new JBIException(ioe);
} // end of try / catch
} // end of if
} // end of startNonNIO()
private void connectNonNIO() throws JBIException {
try {
this.socket = new Socket(this.host, this.port);
this.in = this.socket.getInputStream();
this.out = this.socket.getOutputStream();
this.running = true;
} catch (ConnectException ce) {
log.warn(this.host + ":" + this.port + " unavailable, waiting "
+ this.retryInterval + " milliseconds before trying again.");
this.running = false;
try {
Thread.sleep(this.retryInterval);
} catch (InterruptedException e) {}
this.connectNonNIO();
} catch (Exception e) {
log.error("Exception while trying to make a tcp connection to "
+ this.host + ":" + this.port);
log.error(e);
this.running = false;
throw new JBIException("Failed to connect to " + this.host + ":"
+ this.port, e);
} // end of try / catch
} // end of connectNonNIO()
private void stopNIO() throws JBIException {
log.debug("Inside stopNIO()");
if (this.socketChannel != null) {
this.cleanup();
try {
this.socketChannel.close();
this.socketChannel = null;
this.running = false;
} catch (IOException e) {
throw new JBIException("Unable to close connection", e);
} // end of try / catch
} // end of if
} // end of stopNIO()
private void stopNonNIO() throws JBIException {
log.debug("Inside stopNonNIO()");
if (this.socket != null) {
if (this.socket.isConnected()) {
this.cleanup();
try {
this.socket.close();
this.in = null;
this.out = null;
this.socket = null;
this.running = false;
} catch (IOException e) {
throw new JBIException("Unable to close connection", e);
} // end of try / catch
} // end of if
} // end of if
} // end of stopNonNIO()
} // end of class TcpComponentSupport
On 5/12/06 11:49 AM, "Doug Fischer" <[EMAIL PROTECTED]> wrote:
> Sorry, let me try it again with just the .java file.
>
>
> On 5/12/06 11:47 AM, "Doug Fischer" <[EMAIL PROTECTED]> wrote:
>
>> Hopefully this time the attachment will go through. I am not exactly sure
>> why it didn't before. It is a zip file.
>>
>> I guess I was thinking that when the component is deployed into the
>> lwcontainer that it would be running in its own thread anyway. Apparently I
>> was incorrect in my assumption.
>>
>> I am still trying to get a grasp on the whole ServiceMix thing so please
>> excuse my ignorance.
>>
>> Thanks,
>> Doug
>>
>>
>> On 5/12/06 11:19 AM, "Guillaume Nodet" <[EMAIL PROTECTED]> wrote:
>>
>>> I think the attachment have been lost.
>>>
>>> Btw, for a tcp consumer that creates a server socket, it has to wait
>>> for incoming connections, so how can you not start a thread ? The
>>> jabber api uses a listener, so it is really different.
>>>
>>> Cheers,
>>> Guillaume Nodet
>>>
>>> On 5/12/06, Doug Fischer <[EMAIL PROTECTED]> wrote:
>>>> I have attached my TcpComponentSupport class to this email. If you could
>>>> be
>>>> so kind to take a look at it and let me know if it looks ok. You mentioned
>>>> that the start method should start a Thread and then kill that Thread in
>>>> the
>>>> stop method. I am not actually doing this right now however I of course
>>>> could implement it that way if necessary. I attempted to use the
>>>> JabberComponentSupport class as a template.
>>>>
>>>> Please let me know your thoughts.
>>>>
>>>> Thanks,
>>>> Doug
>>>>
>>>>
>>>> On 5/12/06 8:06 AM, "Guillaume Nodet" <[EMAIL PROTECTED]> wrote:
>>>>
>>>>> Not sure to understand the problem.
>>>>> Usually a lightweight consumer BC will begin to receive messages when
>>>>> the component is started. It means that all the necessary
>>>>> configuration parameters have been set.
>>>>> The tcp component needs to start a thread inside the start method (and
>>>>> kill it inside the stop method).
>>>>>
>>>>> The second problem is related to components / service units lifecycle.
>>>>> When a lightweight component is deployed to the lwcontainer, he has a
>>>>> specific lifecycle: when the service unit is started, the component is
>>>>> started and when the service unit is stopped, the component is
>>>>> shutdown.
>>>>>
>>>>> But the problem should only occur when someone sends something on the
>>>>> socket
>>>>> ...
>>>>>
>>>>> Cheers,
>>>>> Guillaume Nodet
>>>>>
>>>>> On 5/12/06, Doug Fischer <[EMAIL PROTECTED]> wrote:
>>>>>> Guillaume,
>>>>>>
>>>>>> Yes you are right, I did ask before (somehow I got sidetracked an forgot
>>>>>> all
>>>>>> about it). Anyway, thank you for the reply though.
>>>>>>
>>>>>> I have been able to implement a TcpComponentSupport class, at least I
>>>>>> think
>>>>>> so anyway. The problem that I am having now is that as soon as I deploy
>>>>>> the
>>>>>> component it starts collecting data and pushing it to the bus. I have
>>>>>> run
>>>>>> into a couple of problems because of this. The first problem is while
>>>>>> testing, I would like to extend from TestSupport and use the
>>>>>> assertMessagesReceived(), however when it loads the ApplicationContext,
>>>>>> the
>>>>>> concrete TcpComponentSupport class simply begins communicating over the
>>>>>> socket and the test method with the assertMessagesReceived() call is
>>>>>> never
>>>>>> executed. The test will just run forever.
>>>>>>
>>>>>> The second question that I have is along the same lines. I would like to
>>>>>> create a service unit, with simply this component in it and deploy it to
>>>>>> ServiceMix and then use something like eip to route the messages where I
>>>>>> want them. The same kind of problem occurs in this scenario, as soon as
>>>>>> the
>>>>>> service unit is deployed, it will start collecting data from the stream
>>>>>> and
>>>>>> trying to push it to the bus, the problem is that there is no destination
>>>>>> assigned to the the component yet so it tries to send it to "service:
>>>>>> null
>>>>>> and interface null".
>>>>>>
>>>>>> Do you have any ideas?
>>>>>>
>>>>>> Thank you,
>>>>>> Doug
>>>>>>
>>>>>>
>>>>>> On 5/11/06 3:19 AM, "Guillaume Nodet" <[EMAIL PROTECTED]> wrote:
>>>>>>
>>>>>>> It seems there's none -- if I recall, you have already asked this
>>>>>>> question
>>>>>>> ;)
>>>>>>> However, it should not be very difficult to implement.
>>>>>>> You have to inherit the
>>>>>>> org.apache.servicemix.components.util.ComponentSupport and implements
>>>>>>> MessageExchangeListener. Then, just create your socket and read it :)
>>>>>>> You may need another lw component for sending over tcp, also.
>>>>>>>
>>>>>>> Cheers,
>>>>>>> Guillaume Nodet
>>>>>>>
>>>>>>> On 5/10/06, Doug Fischer <[EMAIL PROTECTED]> wrote:
>>>>>>>> Does anyone know if a basic TCP component is available? What I would
>>>>>>>> like
>>>>>>>> to be able to do is simply extend from a TCP component, set the host
>>>>>>>> and
>>>>>>>> port in the servicemix.xml file, and have access to the input and
>>>>>>>> output
>>>>>>>> streams in my implementation class. I would then want to be able to
>>>>>>>> read
>>>>>>>> from the input stream and create a NormalizedMessage to put on the bus
>>>>>>>> and
>>>>>>>> also be able to get a message from the bus and write it to the output
>>>>>>>> stream.
>>>>>>>>
>>>>>>>> Is there anything like this out there?
>>>>>>>>
>>>>>>>> Thanks,
>>>>>>>> Doug
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>>>
>>>>
>>>
>>
>
