1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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.qpid.example.subscriber;
import org.apache.log4j.Logger;
import org.apache.qpid.example.shared.Statics;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.JMSException;
import javax.jms.Queue;
/**
* Subclass of Subscriber which consumes a heartbeat message
* Author: Marnie McCormack
* Date: 12-Sep-2006
* Time: 09:41:07
* Copyright JPMorgan Chase 2006
*/
public class MonitoredSubscriber extends Subscriber
{
private static final Logger _logger = Logger.getLogger(MonitoredSubscriber.class);
private static MessageConsumer _monitorConsumer;
public static class MonitorMessageListener implements MessageListener
{
private String _name;
public MonitorMessageListener(String name)
{
_name = name;
}
/*
* Listens for heartbeat messages and acknowledges them
*/
public void onMessage(javax.jms.Message message)
{
_logger.info(_name + " monitor got message '" + message + "'");
try
{
_logger.debug("Monitor acknowledging recieved message");
//Now acknowledge the message to clear it from our queue
message.acknowledge();
}
catch(JMSException j)
{
_logger.error("Monitor caught JMSException trying to acknowledge message receipt");
j.printStackTrace();
}
catch(Exception e)
{
_logger.error("Monitor caught unexpected exception trying to handle message");
e.printStackTrace();
}
}
}
/*
* Subscribes to Queue and attaches additional monitor listener
* @param hostdetails - for broker connection in host1:port1;host2:port2 format
* @param username - for connection to the broker
* @password - for connection to the broker
* @virtualpath
*/
public void subscribeAndMonitor(String hostdetails, String username, String password,
String virtualPath, String queueName)
{
Queue queue;
try
{
//Create monitor comsumer for failover purposes
if (queueName==null||queueName.length()==0)
{
queue = getSession(_connection).createQueue(Statics.QUEUE_NAME);
}
else
{
queue = getSession(_connection).createQueue(queueName);
}
_monitorConsumer = getSession(_connection).createConsumer(queue);
//give the monitor message listener a name of it's own
_monitorConsumer.setMessageListener(new MonitoredSubscriber.MonitorMessageListener("MonitorListener " + System.currentTimeMillis()));
MonitoredSubscriber._logger.info("Starting monitored subscription ...");
MonitoredSubscriber._connection.start();
//and now start ordinary consumption too
subscribe(hostdetails,username,password,virtualPath,queueName);
}
catch (Throwable t)
{
_logger.error("Fatal error: " + t);
t.printStackTrace();
}
}
//stop consuming
public void stopMonitor()
{
try
{
_monitorConsumer.close();
_monitorConsumer = null;
stop();
}
catch(JMSException j)
{
_logger.error("JMSException trying to Subscriber.stop: " + j.getStackTrace());
}
}
}
|
