You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by Juleian <ju...@gmx.net> on 2017/08/18 17:26:16 UTC
Close message listener decoupled
We want to use ActiveMQ within a tool for JMS communication. I wrote a class
containing a method that handles the subscription to the broker with a
certain topic. So by using my register(topic) method, a message listener is
created that listens to the topic mentioned in the parameter. This register
mehtod will be used by java classes within our tool. Now I need to be able
to close those message listeners when the component (java class) wants to.
But I dont see a way by now, how to implement a "unsubscribe" method.
So in a nutshell, I want a java class to be able to call register(test); and
when it does not want to receiver messages anymore, call
unsubscribe(parameter necessary?) to close the session. How can I adress the
ceated session and close it within my java class?
Can somebody help me?
--
View this message in context: http://activemq.2283324.n4.nabble.com/Close-message-listener-decoupled-tp4729814.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: Close message listener decoupled
Posted by Tim Bain <tb...@alumni.duke.edu>.
One other problem with the implementation: you provided a MessageListener
of your own definition, which doesn't do anything, which means your
consumer has no way to get the messages. A better alternative would be to
have your register method take the MessageListener object (of the
consumer's definition, not your own) as an argument.
Tim
On Aug 18, 2017 1:02 PM, "Justin Bertram" <jb...@redhat.com> wrote:
> > I can't see any reason why anyone would ever not want to use ActiveMQ ;)
>
> I agree. However, based on his code he's using 5.x right now, and I can
> imagine a day when he'd want to move to ActiveMQ Artemis. :)
>
>
> Justin
>
> On Fri, Aug 18, 2017 at 1:59 PM, Timothy Bish <ta...@gmail.com> wrote:
>
> > On 08/18/2017 02:49 PM, Justin Bertram wrote:
> >
> >> I see a couple of potential problems with this code:
> >>
> >> 1) It's littered with non-JMS code (e.g. ActiveMQConnectionFactory,
> >> ActiveMQConnection). You may not care about that if this application
> will
> >> only ever integrate with ActiveMQ, but if you think you'll ever want to
> >> use
> >> any other kind of JMS broker then you'll have to rewrite some of this
> >> stuff
> >> when you migrate. APIs are standardized for a reason. Deviate at your
> >> own
> >> risk.
> >>
> >
> > I can't see any reason why anyone would ever not want to use ActiveMQ ;)
> >
> >
> >
> >> 2) Every time you invoke register() you create a new Connection,
> Session,
> >> Consumer, etc. This is an anti-pattern. Connection objects are "heavy"
> >> and are meant to be shared.
> >>
> >> 3) You do not keep track of any of the resources you create (i.e.
> >> Connection, Session, Consumer) so you have no way to clean them up.
> >>
> >> > maybe the strcture is not good at all
> >>
> >> I'd agree with that.
> >>
> >>
> >> Justin
> >>
> >> On Fri, Aug 18, 2017 at 12:59 PM, Juleian <ju...@gmx.net>
> wrote:
> >>
> >> That sounds reasonable actually. The only problem I also somehow have is
> >>> when
> >>> register is called the method "is stuck" in the thread that is opened
> or
> >>> not? I do not see how I can return a value with this structure (maybe
> the
> >>> strcture is not good at all, its my first real programming task).
> >>>
> >>> public register(String topicName) {
> >>>
> >>>
> >>>
> >>>
> >>>
> >>> try {
> >>>
> >>> // create connection to Broker, create Session
> >>> and
> >>> Consumer
> >>>
> >>> ActiveMQConnectionFactory connectionFactory =
> >>> new
> >>> ActiveMQConnectionFactory(
> >>> "tcp://localhost:61616");
> >>> ActiveMQConnection connection =
> >>> (ActiveMQConnection)
> >>> connectionFactory.createConnection();
> >>> connection.start();
> >>> Session session =
> connection.createSession(false
> >>> ,
> >>> Session.AUTO_ACKNOWLEDGE);
> >>> Topic topicObject =
> >>> session.createTopic(topicName)
> >>> ;
> >>> MessageConsumer consumer =
> >>> session.createConsumer(
> >>> topicObject);
> >>>
> >>>
> >>> //check if topic is available on Broker
> >>>
> >>>
> >>>
> >>> DestinationSource ds = connection.
> >>> getDestinationSource();
> >>>
> >>> Set<ActiveMQTopic> topics = ds.getTopics();
> >>> String compare = topicName;
> >>> int count = 0;
> >>>
> >>> for(ActiveMQTopic topic : topics){
> >>>
> >>>
> >>>
> >>> if(compare.equals(topic.getTopicName()))
> {
> >>>
> >>> System.out.println("Found " +
> >>> topic.getTopicName());
> >>> count = count + 1;
> >>>
> >>>
> >>> }
> >>>
> >>> }
> >>>
> >>> if(count == 0){
> >>>
> >>>
> >>> System.out.println("The topic you want
> >>> to
> >>> subscribe to is not
> >>> found.");
> >>> System.out.println("Please try again
> >>> with
> >>> a valid topic name.");
> >>>
> >>> return;
> >>>
> >>> }
> >>>
> >>>
> >>>
> >>>
> >>> //register Component at mongoDB
> >>>
> >>>
> >>>
> >>> MessageListener listner = new
> MessageListener()
> >>> {
> >>>
> >>> @Override
> >>> public void onMessage(Message
> message)
> >>> {
> >>> try {
> >>> if (message instanceof TextMessage) {
> >>> TextMessage textMessage = (TextMessage)
> >>> message;
> >>> System.out.println("Received message : "
> >>> + textMessage.getText() + "'");
> >>> }
> >>> } catch (JMSException e) {
> >>> System.out.println("Caught:" + e);
> >>> }
> >>> }
> >>> };
> >>> consumer.setMessageListener(listner);
> >>>
> >>> try {
> >>> System.in.read();
> >>> } catch (IOException e) {
> >>> }
> >>>
> >>> } catch (JMSException ex) {
> >>> // Logger.getLogger(Consumer.class.getName()).log(Level.
> >>> SEVERE,
> >>> null, ex);
> >>>
> >>> }
> >>>
> >>>
> >>>
> >>> }
> >>>
> >>>
> >>>
> >>>
> >>> --
> >>> View this message in context: http://activemq.2283324.n4.
> >>> nabble.com/Close-message-listener-decoupled-tp4729814p4729816.html
> >>> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
> >>>
> >>>
> > --
> > Tim Bish
> > twitter: @tabish121
> > blog: http://timbish.blogspot.com/
> >
> >
>
Re: Close message listener decoupled
Posted by Justin Bertram <jb...@redhat.com>.
> I can't see any reason why anyone would ever not want to use ActiveMQ ;)
I agree. However, based on his code he's using 5.x right now, and I can
imagine a day when he'd want to move to ActiveMQ Artemis. :)
Justin
On Fri, Aug 18, 2017 at 1:59 PM, Timothy Bish <ta...@gmail.com> wrote:
> On 08/18/2017 02:49 PM, Justin Bertram wrote:
>
>> I see a couple of potential problems with this code:
>>
>> 1) It's littered with non-JMS code (e.g. ActiveMQConnectionFactory,
>> ActiveMQConnection). You may not care about that if this application will
>> only ever integrate with ActiveMQ, but if you think you'll ever want to
>> use
>> any other kind of JMS broker then you'll have to rewrite some of this
>> stuff
>> when you migrate. APIs are standardized for a reason. Deviate at your
>> own
>> risk.
>>
>
> I can't see any reason why anyone would ever not want to use ActiveMQ ;)
>
>
>
>> 2) Every time you invoke register() you create a new Connection, Session,
>> Consumer, etc. This is an anti-pattern. Connection objects are "heavy"
>> and are meant to be shared.
>>
>> 3) You do not keep track of any of the resources you create (i.e.
>> Connection, Session, Consumer) so you have no way to clean them up.
>>
>> > maybe the strcture is not good at all
>>
>> I'd agree with that.
>>
>>
>> Justin
>>
>> On Fri, Aug 18, 2017 at 12:59 PM, Juleian <ju...@gmx.net> wrote:
>>
>> That sounds reasonable actually. The only problem I also somehow have is
>>> when
>>> register is called the method "is stuck" in the thread that is opened or
>>> not? I do not see how I can return a value with this structure (maybe the
>>> strcture is not good at all, its my first real programming task).
>>>
>>> public register(String topicName) {
>>>
>>>
>>>
>>>
>>>
>>> try {
>>>
>>> // create connection to Broker, create Session
>>> and
>>> Consumer
>>>
>>> ActiveMQConnectionFactory connectionFactory =
>>> new
>>> ActiveMQConnectionFactory(
>>> "tcp://localhost:61616");
>>> ActiveMQConnection connection =
>>> (ActiveMQConnection)
>>> connectionFactory.createConnection();
>>> connection.start();
>>> Session session = connection.createSession(false
>>> ,
>>> Session.AUTO_ACKNOWLEDGE);
>>> Topic topicObject =
>>> session.createTopic(topicName)
>>> ;
>>> MessageConsumer consumer =
>>> session.createConsumer(
>>> topicObject);
>>>
>>>
>>> //check if topic is available on Broker
>>>
>>>
>>>
>>> DestinationSource ds = connection.
>>> getDestinationSource();
>>>
>>> Set<ActiveMQTopic> topics = ds.getTopics();
>>> String compare = topicName;
>>> int count = 0;
>>>
>>> for(ActiveMQTopic topic : topics){
>>>
>>>
>>>
>>> if(compare.equals(topic.getTopicName())) {
>>>
>>> System.out.println("Found " +
>>> topic.getTopicName());
>>> count = count + 1;
>>>
>>>
>>> }
>>>
>>> }
>>>
>>> if(count == 0){
>>>
>>>
>>> System.out.println("The topic you want
>>> to
>>> subscribe to is not
>>> found.");
>>> System.out.println("Please try again
>>> with
>>> a valid topic name.");
>>>
>>> return;
>>>
>>> }
>>>
>>>
>>>
>>>
>>> //register Component at mongoDB
>>>
>>>
>>>
>>> MessageListener listner = new MessageListener()
>>> {
>>>
>>> @Override
>>> public void onMessage(Message message)
>>> {
>>> try {
>>> if (message instanceof TextMessage) {
>>> TextMessage textMessage = (TextMessage)
>>> message;
>>> System.out.println("Received message : "
>>> + textMessage.getText() + "'");
>>> }
>>> } catch (JMSException e) {
>>> System.out.println("Caught:" + e);
>>> }
>>> }
>>> };
>>> consumer.setMessageListener(listner);
>>>
>>> try {
>>> System.in.read();
>>> } catch (IOException e) {
>>> }
>>>
>>> } catch (JMSException ex) {
>>> // Logger.getLogger(Consumer.class.getName()).log(Level.
>>> SEVERE,
>>> null, ex);
>>>
>>> }
>>>
>>>
>>>
>>> }
>>>
>>>
>>>
>>>
>>> --
>>> View this message in context: http://activemq.2283324.n4.
>>> nabble.com/Close-message-listener-decoupled-tp4729814p4729816.html
>>> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>>>
>>>
> --
> Tim Bish
> twitter: @tabish121
> blog: http://timbish.blogspot.com/
>
>
Re: Close message listener decoupled
Posted by Timothy Bish <ta...@gmail.com>.
On 08/18/2017 02:49 PM, Justin Bertram wrote:
> I see a couple of potential problems with this code:
>
> 1) It's littered with non-JMS code (e.g. ActiveMQConnectionFactory,
> ActiveMQConnection). You may not care about that if this application will
> only ever integrate with ActiveMQ, but if you think you'll ever want to use
> any other kind of JMS broker then you'll have to rewrite some of this stuff
> when you migrate. APIs are standardized for a reason. Deviate at your own
> risk.
I can't see any reason why anyone would ever not want to use ActiveMQ ;)
>
> 2) Every time you invoke register() you create a new Connection, Session,
> Consumer, etc. This is an anti-pattern. Connection objects are "heavy"
> and are meant to be shared.
>
> 3) You do not keep track of any of the resources you create (i.e.
> Connection, Session, Consumer) so you have no way to clean them up.
>
> > maybe the strcture is not good at all
>
> I'd agree with that.
>
>
> Justin
>
> On Fri, Aug 18, 2017 at 12:59 PM, Juleian <ju...@gmx.net> wrote:
>
>> That sounds reasonable actually. The only problem I also somehow have is
>> when
>> register is called the method "is stuck" in the thread that is opened or
>> not? I do not see how I can return a value with this structure (maybe the
>> strcture is not good at all, its my first real programming task).
>>
>> public register(String topicName) {
>>
>>
>>
>>
>>
>> try {
>>
>> // create connection to Broker, create Session and
>> Consumer
>>
>> ActiveMQConnectionFactory connectionFactory = new
>> ActiveMQConnectionFactory(
>> "tcp://localhost:61616");
>> ActiveMQConnection connection =
>> (ActiveMQConnection)
>> connectionFactory.createConnection();
>> connection.start();
>> Session session = connection.createSession(false,
>> Session.AUTO_ACKNOWLEDGE);
>> Topic topicObject = session.createTopic(topicName)
>> ;
>> MessageConsumer consumer = session.createConsumer(
>> topicObject);
>>
>>
>> //check if topic is available on Broker
>>
>>
>>
>> DestinationSource ds = connection.
>> getDestinationSource();
>>
>> Set<ActiveMQTopic> topics = ds.getTopics();
>> String compare = topicName;
>> int count = 0;
>>
>> for(ActiveMQTopic topic : topics){
>>
>>
>>
>> if(compare.equals(topic.getTopicName())) {
>>
>> System.out.println("Found " +
>> topic.getTopicName());
>> count = count + 1;
>>
>>
>> }
>>
>> }
>>
>> if(count == 0){
>>
>>
>> System.out.println("The topic you want to
>> subscribe to is not
>> found.");
>> System.out.println("Please try again with
>> a valid topic name.");
>>
>> return;
>>
>> }
>>
>>
>>
>>
>> //register Component at mongoDB
>>
>>
>>
>> MessageListener listner = new MessageListener() {
>>
>> @Override
>> public void onMessage(Message message) {
>> try {
>> if (message instanceof TextMessage) {
>> TextMessage textMessage = (TextMessage)
>> message;
>> System.out.println("Received message : "
>> + textMessage.getText() + "'");
>> }
>> } catch (JMSException e) {
>> System.out.println("Caught:" + e);
>> }
>> }
>> };
>> consumer.setMessageListener(listner);
>>
>> try {
>> System.in.read();
>> } catch (IOException e) {
>> }
>>
>> } catch (JMSException ex) {
>> // Logger.getLogger(Consumer.class.getName()).log(Level.
>> SEVERE,
>> null, ex);
>>
>> }
>>
>>
>>
>> }
>>
>>
>>
>>
>> --
>> View this message in context: http://activemq.2283324.n4.
>> nabble.com/Close-message-listener-decoupled-tp4729814p4729816.html
>> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>>
--
Tim Bish
twitter: @tabish121
blog: http://timbish.blogspot.com/
Re: Close message listener decoupled
Posted by Juleian <ju...@gmx.net>.
Thank you guys for the input. I will think about using only JMS related
statements.
I kinda get your point of antipattern and lack of tracking. Since I am not
an experienced programmer
(this is task of my master thesis), is there anythign out there on the net
that can help me with designing a better solution? I feel a bid lost in
terms of documentation. I had a look at "ActiveMQ in Action" and Oreillys
JMS in general, but this does no seem to help me that much for my use case.
Best regards !
--
View this message in context: http://activemq.2283324.n4.nabble.com/Close-message-listener-decoupled-tp4729814p4729859.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: Close message listener decoupled
Posted by Justin Bertram <jb...@redhat.com>.
I see a couple of potential problems with this code:
1) It's littered with non-JMS code (e.g. ActiveMQConnectionFactory,
ActiveMQConnection). You may not care about that if this application will
only ever integrate with ActiveMQ, but if you think you'll ever want to use
any other kind of JMS broker then you'll have to rewrite some of this stuff
when you migrate. APIs are standardized for a reason. Deviate at your own
risk.
2) Every time you invoke register() you create a new Connection, Session,
Consumer, etc. This is an anti-pattern. Connection objects are "heavy"
and are meant to be shared.
3) You do not keep track of any of the resources you create (i.e.
Connection, Session, Consumer) so you have no way to clean them up.
> maybe the strcture is not good at all
I'd agree with that.
Justin
On Fri, Aug 18, 2017 at 12:59 PM, Juleian <ju...@gmx.net> wrote:
> That sounds reasonable actually. The only problem I also somehow have is
> when
> register is called the method "is stuck" in the thread that is opened or
> not? I do not see how I can return a value with this structure (maybe the
> strcture is not good at all, its my first real programming task).
>
> public register(String topicName) {
>
>
>
>
>
> try {
>
> // create connection to Broker, create Session and
> Consumer
>
> ActiveMQConnectionFactory connectionFactory = new
> ActiveMQConnectionFactory(
> "tcp://localhost:61616");
> ActiveMQConnection connection =
> (ActiveMQConnection)
> connectionFactory.createConnection();
> connection.start();
> Session session = connection.createSession(false,
> Session.AUTO_ACKNOWLEDGE);
> Topic topicObject = session.createTopic(topicName)
> ;
> MessageConsumer consumer = session.createConsumer(
> topicObject);
>
>
> //check if topic is available on Broker
>
>
>
> DestinationSource ds = connection.
> getDestinationSource();
>
> Set<ActiveMQTopic> topics = ds.getTopics();
> String compare = topicName;
> int count = 0;
>
> for(ActiveMQTopic topic : topics){
>
>
>
> if(compare.equals(topic.getTopicName())) {
>
> System.out.println("Found " +
> topic.getTopicName());
> count = count + 1;
>
>
> }
>
> }
>
> if(count == 0){
>
>
> System.out.println("The topic you want to
> subscribe to is not
> found.");
> System.out.println("Please try again with
> a valid topic name.");
>
> return;
>
> }
>
>
>
>
> //register Component at mongoDB
>
>
>
> MessageListener listner = new MessageListener() {
>
> @Override
> public void onMessage(Message message) {
> try {
> if (message instanceof TextMessage) {
> TextMessage textMessage = (TextMessage)
> message;
> System.out.println("Received message : "
> + textMessage.getText() + "'");
> }
> } catch (JMSException e) {
> System.out.println("Caught:" + e);
> }
> }
> };
> consumer.setMessageListener(listner);
>
> try {
> System.in.read();
> } catch (IOException e) {
> }
>
> } catch (JMSException ex) {
> // Logger.getLogger(Consumer.class.getName()).log(Level.
> SEVERE,
> null, ex);
>
> }
>
>
>
> }
>
>
>
>
> --
> View this message in context: http://activemq.2283324.n4.
> nabble.com/Close-message-listener-decoupled-tp4729814p4729816.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
Re: Close message listener decoupled
Posted by Juleian <ju...@gmx.net>.
That sounds reasonable actually. The only problem I also somehow have is when
register is called the method "is stuck" in the thread that is opened or
not? I do not see how I can return a value with this structure (maybe the
strcture is not good at all, its my first real programming task).
public register(String topicName) {
try {
// create connection to Broker, create Session and Consumer
ActiveMQConnectionFactory connectionFactory = new
ActiveMQConnectionFactory(
"tcp://localhost:61616");
ActiveMQConnection connection = (ActiveMQConnection)
connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
Topic topicObject = session.createTopic(topicName);
MessageConsumer consumer = session.createConsumer(topicObject);
//check if topic is available on Broker
DestinationSource ds = connection.getDestinationSource();
Set<ActiveMQTopic> topics = ds.getTopics();
String compare = topicName;
int count = 0;
for(ActiveMQTopic topic : topics){
if(compare.equals(topic.getTopicName())) {
System.out.println("Found " + topic.getTopicName());
count = count + 1;
}
}
if(count == 0){
System.out.println("The topic you want to subscribe to is not
found.");
System.out.println("Please try again with a valid topic name.");
return;
}
//register Component at mongoDB
MessageListener listner = new MessageListener() {
@Override
public void onMessage(Message message) {
try {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
System.out.println("Received message : "
+ textMessage.getText() + "'");
}
} catch (JMSException e) {
System.out.println("Caught:" + e);
}
}
};
consumer.setMessageListener(listner);
try {
System.in.read();
} catch (IOException e) {
}
} catch (JMSException ex) {
// Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE,
null, ex);
}
}
--
View this message in context: http://activemq.2283324.n4.nabble.com/Close-message-listener-decoupled-tp4729814p4729816.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: Close message listener decoupled
Posted by Justin Bertram <jb...@redhat.com>.
The problem you've described is pretty vague and open-ended in my opinion.
It's hard to give any advice in such a situation, but here's a thought...
You could have register() return the javax.jms.MessageConsumer on which the
message listener is set so when you want to close it you can just call
MessageConsumer.close().
Justin
On Fri, Aug 18, 2017 at 12:26 PM, Juleian <ju...@gmx.net> wrote:
> We want to use ActiveMQ within a tool for JMS communication. I wrote a
> class
> containing a method that handles the subscription to the broker with a
> certain topic. So by using my register(topic) method, a message listener is
> created that listens to the topic mentioned in the parameter. This register
> mehtod will be used by java classes within our tool. Now I need to be able
> to close those message listeners when the component (java class) wants to.
> But I dont see a way by now, how to implement a "unsubscribe" method.
>
> So in a nutshell, I want a java class to be able to call register(test);
> and
> when it does not want to receiver messages anymore, call
> unsubscribe(parameter necessary?) to close the session. How can I adress
> the
> ceated session and close it within my java class?
>
> Can somebody help me?
>
>
>
>
>
> --
> View this message in context: http://activemq.2283324.n4.
> nabble.com/Close-message-listener-decoupled-tp4729814.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>