| |
|
RSS Feed - WebSphere MQ Support
|
RSS Feed - Message Broker Support
|
  |
|
| Stopping a java compute node |
« View previous topic :: View next topic » |
| Author |
Message
|
| rsandoz |
 Posted: Fri Jan 18, 2008 12:09 pm Post subject: Stopping a java compute node |
 |
|
Novice
Joined: 12 Oct 2006
Posts: 17
|
I have been looking for a way to recieve the "stop message flow" call from MessageBroker v6.
The problem is that I am creating an MbJavaComputeNode will run user defined SQL that may not be tuned right and hence run for 10 hours as opposed to 10 seconds.
In our environment, we have to be able to stop the message flow without bouncing the broker. Resetting the broker is the only way to recover from the above scenario.
| Code: |
public void evaluate(MbMessageAssembly contact admin) throws MbException {
try {
for (int i = 0; i < 100; i++) {
Thread.sleep(1000); // I know people here don't like sleep in JCNs
// as I am just using it to simulate long-running java code
System.out.println("Still running");
}
} catch (Exception e) {
e.printStackTrace();
}
}
|
I actually found a way around this problem. Start another thread to monitor the broker queue. From here, I intercept the stop message, set a boolean to stop my java process, then resubmit the stop message so the flow actually stops.
| Code: |
public void evaluate(MbMessageAssembly contact admin) throws MbException {
try {
startMonitor(getBroker().getQueueManagerName());
for (int i = 0; i < 100; i++) {
Thread.sleep(1000); // I know people here don't like sleep in JCNs
// as I am just using it to simulate long-running java code
System.out.println("Still running");
}
} catch (Exception e) {
e.printStackTrace();
}
}
private void startMonitor(final String qmgr) {
new Thread() { public void run() {
try {
String hostname = "LOCALHOST";
int qMgrPort = 1414;
String channel = "SYSTEM.DEF.SVRCONN";
String queue = "SYSTEM.BROKER.ADMIN.QUEUE";
for(;;) {
MQEnvironment.hostname = hostname;
MQEnvironment.port = qMgrPort;
MQEnvironment.channel = channel;
// intercept the message
MQQueueManager qManager = new MQQueueManager(qmgr);
MQQueue mqqueue = qManager.accessQueue(queue
, MQC.MQOO_FAIL_IF_QUIESCING | MQC.MQOO_SAVE_ALL_CONTEXT | MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT
, null, null, null);
MQMessage mqmessage = new MQMessage();
mqmessage.correlationId = MQC.MQCI_NONE;
MQGetMessageOptions mqgetmessageoptions = new MQGetMessageOptions();
mqgetmessageoptions.options = MQC.MQMO_MATCH_MSG_ID | MQC.MQGMO_FAIL_IF_QUIESCING;
mqgetmessageoptions.waitInterval = -1;
mqqueue.get(mqmessage, mqgetmessageoptions);
String strMsg = mqmessage.readLine();
// put the message back
MQPutMessageOptions mqputmessageoptions = new MQPutMessageOptions();
mqputmessageoptions.options = 2;
mqqueue.put(mqmessage);
mqqueue.close();
qManager.disconnect();
// need to do a better job of parsing here
if (strMsg.indexOf("<Stop>")>-1)
break;
}
System.out.println("Received stop");
MyFlow.stopped = true;
} catch (Exception e) {
System.out.println("::Exception::");
e.printStackTrace();
}
}}.start();
}
|
The problem with this is that my code where I parse the stop message stops for any flow in the execution group. What I need is to make this specific to my message flow. From looking at the contents of strMsg:
| Code: |
<Broker label="MBLOCAL" uuid="9609cd7c-1601-0000-0080-99c93b774709" version="1">
<ExecutionGroup uuid="9709cd7c-1601-0000-0080-99c93b774709">
<Stop>
<MessageFlow uuid="82c5618d-1701-0000-0080-a1ea79c7616e"/>
</Stop>
</ExecutionGroup>
</Broker>
|
I have found that everything is identified by uuids. I can get the uuid for the broker pretty easily (system properties), but the execution group and message flow proved more difficult. The only thing I found was the ConfigManagerProxy api, which in my configuration, required me to set the user id before being able to access it:
| Code: |
System.setProperty("user.name","db2admin");
MQConfigManagerConnectionParameters mqcmcp = new MQConfigManagerConnectionParameters(hostname,cfgMgrPort,qmgr);
mqcmcp.disableDomainAwareness();
ConfigManagerProxy cmp = ConfigManagerProxy.getInstance(mqcmcp);
BrokerProxy bp = cmp.getTopology().getBrokerByName(qmgr);
ExecutionGroupProxy egp = bp.getExecutionGroupByName(executionGroup);
MessageFlowProxy mfp = egp.getMessageFlowByName(messageFlow);
cmp.disconnect();
|
which improved my parse call:
| Code: |
if (strMsg.indexOf("<Stop>")>-1
&& strMsg.indexOf(bp.getProperty("uuid"))>-1
&& strMsg.indexOf(egp.getProperty("uuid"))>-1
&& strMsg.indexOf(mfp.getProperty("uuid"))>-1
)
|
and I can even doubly make sure my flow gets stopped:
Now I am faced with the problem of getting the port and a user name for the config manager dynamically. I also would like to be able to determine the brokers host, port, queue and channel information dynamically.
Any suggestion, criticisms, etc on the above is appreciated. |
|
| Back to top |
|
 |
|
|
|
| |
|
Page 1 of 1 |
|
You cannot post new topics in this forum
You cannot reply to topics in this forum
You cannot edit your posts in this forum
You cannot delete your posts in this forum
You cannot vote in polls in this forum
|
|
|
|
