// TODO: cluster forwarding with multiple destinations:
String txt = "Messages with more than one destinations in a cluster environment is not implemented, only destination '" + destinationArr[ii].toXml() + "' of '" + msgUnit.getLogId() + "' was delivered";
log.warning(txt);
throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, txt);
}
I_Checkpoint cp = glob.getCheckpointPlugin();
if (cp != null)
cp.passingBy(I_Checkpoint.CP_PUBLISH_ACK, msgUnit, null, null);
return publishReturnQos.toXml();
}
/*
if (publishReturnQos != null) {
// Message was forwarded. TODO: How to return multiple publishReturnQos from multiple destinations? !!!
BUGGY: We need to take a clone to not remove the destination of the sent message
publishQos.removeDestination(destinationArr[ii]);
}
*/
}
/*
if (publishQos.getNumDestinations() == 0) { // we are done, all messages where forwarded
return publishReturnQos.toXml();
}
*/
}
// Publish/Subscribe mode (or if PtP had no result)
else { // if (publishQos.isSubscribable()) {
try {
PublishRetQosWrapper ret = glob.getClusterManager().forwardPublish(sessionInfo, msgUnit);
//Thread.currentThread().dumpStack();
if (ret != null) { // Message was forwarded to master cluster
I_Checkpoint cp = glob.getCheckpointPlugin();
if (cp != null)
cp.passingBy(I_Checkpoint.CP_PUBLISH_ACK, msgUnit, null, null);
publishReturnQos = ret.getPublishReturnQos();
if (ret.getNodeMasterInfo().isDirtyRead() == false) {
if (log.isLoggable(Level.FINE)) log.fine("Message " + msgKeyData.getOid() + " forwarded to master " + ret.getNodeMasterInfo().getId() + ", dirtyRead==false nothing more to do");
return publishReturnQos.toXml();
}
// else we publish it locally as well (dirty read!)
}
}
catch (XmlBlasterException e) {
if (e.getErrorCode() == ErrorCode.RESOURCE_CONFIGURATION_PLUGINFAILED) {
this.glob.setUseCluster(false);
}
else {
e.printStackTrace();
throw e;
}
}
}
}
else {
if (! publishQos.isFromPersistenceStore()) {
if (msgKeyData.isInternal()) {
if (log.isLoggable(Level.FINE)) log.fine("Cluster manager is not ready, handling message '" + msgKeyData.getOid() + "' locally");
}
else {
log.warning("Cluster manager is not ready, handling message '" + msgKeyData.getOid() + "' locally");
}
}
}
}
}
// Handle local message
if (!msgKeyData.getOid().equals(msgUnit.getKeyOid())) {
Thread.dumpStack();
log.severe("Unexpected change of keyOid " + msgKeyData.getOid() + " and msgUnit " + msgUnit.toXml());
}
/*
// Find or create the topic
TopicHandler topicHandler = null;
synchronized(this.topicHandlerMap) {
if (!msgKeyData.getOid().equals(msgUnit.getKeyOid())) {
Thread.dumpStack();
log.severe("Unexpected change of keyOid " + msgKeyData.getOid() + " and msgUnit " + msgUnit.toXml());
}
Object obj = topicHandlerMap.get(msgUnit.getKeyOid());
if (obj == null) {
topicHandler = new TopicHandler(this, sessionInfo, msgUnit.getKeyOid()); // adds itself to topicHandlerMap
}
else {
topicHandler = (TopicHandler)obj;
}
}
*/
TopicHandler topicHandler = null;
try {
topicHandler = this.glob.getTopicAccessor().findOrCreate(sessionInfo, msgUnit.getKeyOid());
// Process the message
publishReturnQos = topicHandler.publish(sessionInfo, msgUnit, publishQos);
}
finally {
this.glob.getTopicAccessor().release(topicHandler);
}
if (publishReturnQos == null) { // assert only
StatusQosData qos = new StatusQosData(glob, MethodName.PUBLISH);
qos.setKeyOid(msgKeyData.getOid());
qos.setState(Constants.STATE_OK);
publishReturnQos = new PublishReturnQos(glob, qos);
publishReturnQos.getData().setRcvTimestamp(publishQos.getRcvTimestamp());
log.severe("Internal: did not excpect to build a PublishReturnQos, but message '" + msgKeyData.getOid() + "' is processed correctly");
Thread.dumpStack();
}
if (!publishQos.isFromPersistenceStore()) {
I_Checkpoint cp = glob.getCheckpointPlugin();
if (cp != null)
cp.passingBy(I_Checkpoint.CP_PUBLISH_ACK, msgUnit, null, null);
}
return publishReturnQos.toXml(); // Use the return value of the cluster master node
}
catch (XmlBlasterException e) {