1919 * MQPCFSubscriber is technically not a subscriber, but a runnable object, which sends PCFCommands every n seconds to
2020 * retrieve specific statistics, which couldn't be retrieved by MQTopicSubscriber.
2121 */
22- public class MQPCFSubscriber implements Runnable {
22+ public class MQPCFSubscriber extends MQSubscriber {
2323 private static final Logger logger = LogManager .getLogger (MQPCFSubscriber .class );
2424 private MQConnection connection ;
2525 private String queueManagerName ;
2626 private MQObject object ;
2727 private PCFMessageAgent agent ;
2828 private List <MQObject > objects ;
29+ private boolean isRunning ;
2930
3031 /**
3132 * MQPCFSubscriber constructor which is used, when exporter is configured to use 1 MQPCFSubscriber per 1 MQObject.
@@ -37,6 +38,7 @@ public class MQPCFSubscriber implements Runnable {
3738 public MQPCFSubscriber (String queueManagerName , Hashtable <String , Object > connectionProperties , MQObject object ) {
3839 establishMQConnection (queueManagerName , connectionProperties );
3940 this .object = object ;
41+ this .isRunning = true ;
4042 }
4143
4244 /**
@@ -50,6 +52,7 @@ public MQPCFSubscriber(String queueManagerName, Hashtable<String, Object> connec
5052 establishMQConnection (queueManagerName , connectionProperties );
5153 this .objects = objects ;
5254 this .object = new MQObject ("*" , objects .get (0 ).getType ());
55+ this .isRunning = true ;
5356 }
5457
5558 /**
@@ -112,19 +115,22 @@ private void updateMetricsWithWildcards(PCFMessage[] pcfResponse) {
112115 /**
113116 * Retrieves info about all objects from input array via direct pcf commands.
114117 *
115- * @param objectNames - input array with objects.
118+ * @param objectNames - input list with objects.
116119 */
117- private void updateWithDirectPCFCommand (ArrayList <String > objectNames ) {
120+ private void updateWithDirectPCFCommand (List <String > objectNames ) {
118121 for (String objectName : objectNames ) {
119122 MQObject directObject = new MQObject (objectName , object .getType ());
120123 try {
121124 PCFMessage [] directPCFResponse = agent .send (directObject .getPcfCmd ());
122125 updateMetricWithoutWildcards (directPCFResponse [0 ], objectName );
123126 } catch (PCFException e ) {
124127 //This error means, that channel has status "inactive".
125- if (e .reasonCode == MQConstants .MQRCCF_CHL_STATUS_NOT_FOUND ) {
128+ if (object . getType () == MQObject . MQType . CHANNEL && e .reasonCode == MQConstants .MQRCCF_CHL_STATUS_NOT_FOUND ) {
126129 logger .warn ("Channel {} is possibly inactive." , objectName );
127130 MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQCHS_INACTIVE ), queueManagerName , objectName );
131+ } else if (object .getType () == MQObject .MQType .LISTENER && e .reasonCode == MQConstants .MQRC_UNKNOWN_OBJECT_NAME ) {
132+ MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQSVC_STATUS_STOPPED ), queueManagerName , objectName );
133+ logger .warn ("Listener {} is possibly stopped." , objectName );
128134 } else {
129135 logger .error ("Error occurred during sending PCF command: " , e );
130136 }
@@ -134,30 +140,46 @@ private void updateWithDirectPCFCommand(ArrayList<String> objectNames) {
134140 }
135141 }
136142
143+ /**
144+ * Stops subscriber.
145+ */
146+ public void stopProcessing () {
147+ isRunning = false ;
148+ try {
149+ agent .disconnect ();
150+ connection .close ();
151+ } catch (MQException e ) {
152+ logger .error ("Error occurred during stopping PCF subscriber: " , e );
153+ }
154+ }
155+
137156 @ Override
138157 public void run () {
139- try {
140- logger .debug ("Sending PCF command for object type {} with name {}..." , object .getType (), object .getName ());
141- PCFMessage [] pcfResponse = agent .send (object .getPcfCmd ());
142- if (!objects .isEmpty ()) {
143- updateMetricsWithWildcards (pcfResponse );
144- } else {
145- for (PCFMessage response : pcfResponse ) {
146- updateMetricWithoutWildcards (response , object .getName ());
158+ if (isRunning ) {
159+ try {
160+ logger .debug ("Sending PCF command for object type {} with name {}..." , object .getType (), object .getName ());
161+ PCFMessage [] pcfResponse = agent .send (object .getPcfCmd ());
162+ if (objects != null && !objects .isEmpty ()) {
163+ updateMetricsWithWildcards (pcfResponse );
164+ } else {
165+ for (PCFMessage response : pcfResponse ) {
166+ updateMetricWithoutWildcards (response , object .getName ());
167+ }
147168 }
169+ logger .debug ("PCF response for object type {} with name {} was processed successfully." , object .getType (), object .getName ());
170+ } catch (PCFException e ) {
171+ if (object .getType () == MQObject .MQType .CHANNEL && e .reasonCode == MQConstants .MQRCCF_CHL_STATUS_NOT_FOUND ) {
172+ logger .warn ("Channel {} is possibly inactive." , object .getName ());
173+ MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQCHS_INACTIVE ), queueManagerName , object .getName ());
174+ } else if (object .getType () == MQObject .MQType .LISTENER && e .reasonCode == MQConstants .MQRC_UNKNOWN_OBJECT_NAME ) {
175+ MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQSVC_STATUS_STOPPED ), queueManagerName , object .getName ());
176+ logger .warn ("Listener {} is possibly stopped." , object .getName ());
177+ } else {
178+ logger .error ("Error occurred during sending PCF command: " , e );
179+ }
180+ } catch (MQException | IOException e ) {
181+ logger .error ("Error occurred during sending PCF command: " , e );
148182 }
149- logger .debug ("PCF response for object type {} with name {} was processed successfully." , object .getType (), object .getName ());
150- } catch (PCFException e ) {
151- if (e .reasonCode == MQConstants .MQRCCF_CHL_STATUS_NOT_FOUND ) {
152- logger .warn ("Channel {} is possibly inactive." , object .getName ());
153- MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQCHS_INACTIVE ), queueManagerName , object .getName ());
154- }
155- if (object .getType () == MQObject .MQType .LISTENER && e .reasonCode == MQConstants .MQRC_UNKNOWN_OBJECT_NAME ) {
156- MetricsManager .updateMetric (MetricsReference .getMetricName (object .getType ()), MetricsReference .getMetricValue (object .getType (), MQConstants .MQSVC_STATUS_STOPPED ), queueManagerName , object .getName ());
157- logger .warn ("Listener {} is possibly stopped." , object .getName ());
158- }
159- } catch (MQException | IOException e ) {
160- logger .error ("Error occurred during sending PCF command: " , e );
161183 }
162184 }
163185
0 commit comments