}
if (log.isLoggable(Level.FINE)) log.fine("get(): Found " + msgUnitList.size() + " remote matches for " + xmlKey.toXml());
}
NEXT_MSG: for (int ii=0; ii<keyDataArr.length; ii++) {
KeyData xmlKeyExact = keyDataArr[ii];
if (xmlKeyExact == null && xmlKey.isExact()) // subscription on a yet unknown message ...
xmlKeyExact = xmlKey;
TopicHandler topicHandler = this.glob.getTopicAccessor().access(xmlKeyExact.getOid());
if( topicHandler == null ) {
/*
if (this.glob.useCluster()) { // cluster support - forward erase to master
try {
MsgUnit tmp[] = glob.getClusterManager().forwardGet(sessionInfo, xmlKey, getQos);
if (tmp != null && tmp.length > 0) {
log.info(ME, "get() access of " + tmp.length + " messages from cluster master");
for (int jj=0; jj<tmp.length; jj++) {
msgUnitList.add(tmp[jj]);
// We currently don' cache the message here in the slave !!!
// We could do it with the xmlBlasterConnection.initCache(int size)
}
continue NEXT_MSG;
}
}
catch (XmlBlasterException e) {
if (e.getErrorCode() == ErrorCode.RESOURCE_CONFIGURATION_PLUGINFAILED) {
this.glob.setUseCluster(false);
}
else {
e.printStackTrace();
throw e;
}
}
}
*/
if (log.isLoggable(Level.FINE)) log.fine("get(): The key '"+xmlKeyExact.getOid()+"' is not available.");
continue NEXT_MSG;
} // topicHandler==null
try {
if (topicHandler.isAlive()) {
int numEntries = getQos.getHistoryQos().getNumEntries();
MsgUnitWrapper[] msgUnitWrapperArr = topicHandler.getMsgUnitWrapperArr(numEntries, getQos.getHistoryQos().getNewestFirst());
NEXT_HISTORY:
for(int kk=0; kk<msgUnitWrapperArr.length; kk++) {
MsgUnitWrapper msgUnitWrapper = msgUnitWrapperArr[kk];
if (msgUnitWrapper == null) {
continue NEXT_HISTORY;
}
if (this.glob.useCluster() && !msgUnitWrapper.getMsgQosData().isAtMaster()) {
if (log.isLoggable(Level.FINE)) log.fine("get(): Ignore message as we are not the master: " + msgUnitWrapper.toXml());
continue NEXT_HISTORY;
}
//topicHandler.checkFilter(SessionInfo publisherSessionInfo, SubscriptionInfo sub, MsgUnitWrapper msgUnitWrapper, boolean handleException)
AccessFilterQos[] filterQos = getQos.getAccessFilterArr();
if (filterQos != null) {
if (log.isLoggable(Level.FINE)) log.fine("Checking " + filterQos.length + " filters");
for (int jj=0; jj<filterQos.length; jj++) {
I_AccessFilter filter = getAccessPluginManager().getAccessFilter(
filterQos[jj].getType(),
filterQos[jj].getVersion(),
msgUnitWrapper.getContentMime(),
msgUnitWrapper.getContentMimeExtended());
if (log.isLoggable(Level.FINE)) log.fine("get("+xmlKeyExact.getOid()+") filter=" + filter + " qos=" + getQos.toXml());
if (filter != null && filter.match(sessionInfo,
msgUnitWrapper.getMsgUnit(),
filterQos[jj].getQuery()) == false)
continue NEXT_HISTORY; // filtered message is not send to client
}