List<SerializedTriple> stl = (List<SerializedTriple>)msg;
// -deserialize the triples and make subjects unique
// -das type-Tripel wird als letztes eingespeist
Literal subject = LiteralFactory.createAnonymousLiteral("_:subj"+UUID.randomUUID());
Triple typeTriple = null;
for(SerializedTriple st : stl) {
Triple t = st.getTriple();
Triple nt = new Triple(subject, t.getPredicate(), t.getObject());
// if this triplet contains the event type, then
// it might be added as a new BProducer
// and is forwarded to appropriate sub brokers
if (t.getPredicate().getName().equals("<http://www.w3.org/1999/02/22-rdf-syntax-ns#type>")){
// first check, if this message came from a producer
// which means that its connection info is not null
if (service.getConnectionInfo() != null){
// add a new BProducer, which only does show effect if
// this specific BProducer did not already exist
this.bModel.addBProducer(new BProducer(t.getObject().getName(), service.getConnectionInfo()));
// now just forward this message
if (this.forwardingTable != null){
List<TcpConnectInfo> forwardAdresses = this.forwardingTable.get(t.getObject().getName());
if (forwardAdresses != null){
for (TcpConnectInfo con : forwardAdresses){
if (!service.getConnectionInfo().equals(con)){
try {
this.connectionTable.get(con).sendMessage(msg);
System.out.println("Forwarded message of type "+t.getObject().getName()+" to "+con.getHost()+":"+con.getPort());
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
}
}
//System.out.println("B:" + nt);
if(nt.getPredicate().compareToNotNecessarilySPARQLSpecificationConform(Literals.RDF.TYPE) == 0) {
typeTriple = nt;
} else {
consumeForAllStreams(nt);
}
}