Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • scs/ipaaca
  • ramin.yaghoubzadeh/ipaaca
2 results
Show changes
Showing
with 1950 additions and 265 deletions
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IURetraction;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.Collection;
import java.util.HashMap;
......@@ -10,6 +45,8 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -17,7 +54,12 @@ import rsb.Event;
import rsb.Factory;
import rsb.Handler;
import rsb.InitializeException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import rsb.Listener;
import rsb.RSBException;
import rsb.Scope;
import rsb.patterns.RemoteServer;
......@@ -25,6 +67,7 @@ import rsb.patterns.RemoteServer;
* An InputBuffer that holds remote IUs.
* @author hvanwelbergen
*/
@Slf4j
public class InputBuffer extends Buffer
{
private Map<String, RemoteServer> remoteServerStore = new HashMap<String, RemoteServer>();
......@@ -34,15 +77,40 @@ public class InputBuffer extends Buffer
private IUStore<RemotePushIU> iuStore = new IUStore<RemotePushIU>();
private IUStore<RemoteMessageIU> messageStore = new IUStore<RemoteMessageIU>();
private String channel = "default";
private boolean resendActive;
public void close()
{
for (Listener listener : listenerStore.values())
{
listener.deactivate();
try
{
listener.deactivate();
}
catch (RSBException e)
{
log.warn("RSB Exception on deactive {}", e, listener.toString());
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
for (RemoteServer remServer : remoteServerStore.values())
{
remServer.deactivate();
try
{
remServer.deactivate();
}
catch (RSBException e)
{
log.warn("RSB Exception on RemoteServer deactivate {}", e, remServer.toString());
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
}
......@@ -63,14 +131,53 @@ public class InputBuffer extends Buffer
// for cat in category_interests:
// self._create_category_listener_if_needed(cat)
public InputBuffer(String owningComponentName, Set<String> categoryInterests)
{
this(owningComponentName, categoryInterests, "default");
}
public InputBuffer(String owningComponentName, Set<String> categoryInterests, String ipaaca_channel)
{
super(owningComponentName);
uniqueName = "/ipaaca/component/" + owningComponentName + "ID" + uuid + "/IB";
resendActive = false;
String shortIDName = getUniqueShortName();
uniqueName = "/ipaaca/component/" + shortIDName + "/IB";
this.channel = ipaaca_channel;
for (String cat : categoryInterests)
{
createCategoryListenerIfNeeded(cat);
}
// add own uuid as identifier for hidden channel. (dlw)
createCategoryListenerIfNeeded(shortIDName);
}
/** Pass resendActive to toggle resendRequest-functionality. */
public InputBuffer(BufferConfiguration bufferconfiguration)
{
super(bufferconfiguration.getOwningComponentName());
this.resendActive = bufferconfiguration.getResendActive();
String shortIDName = getUniqueShortName();
uniqueName = "/ipaaca/component/" + shortIDName + "/IB";
for (String cat : bufferconfiguration.getCategoryInterests())
{
createCategoryListenerIfNeeded(cat);
}
this.channel = bufferconfiguration.getChannel();
// add own uuid as identifier for hidden channel. (dlw)
createCategoryListenerIfNeeded(shortIDName);
}
public boolean isResendActive() {
return this.resendActive;
}
public void setResendActive(boolean active) {
this.resendActive = active;
}
// def _get_remote_server(self, iu):
......@@ -80,7 +187,7 @@ public class InputBuffer extends Buffer
// remote_server = rsb.createRemoteServer(rsb.Scope(str(iu.owner_name)))
// self._remote_server_store[iu.owner_name] = remote_server
// return remote_server
public RemoteServer getRemoteServer(AbstractIU iu)
protected RemoteServer getRemoteServer(AbstractIU iu)
{
if (remoteServerStore.containsKey(iu.getOwnerName()))
{
......@@ -96,10 +203,38 @@ public class InputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
remoteServerStore.put(iu.getOwnerName(), remoteServer);
return remoteServer;
}
protected RemoteServer getRemoteServer(String ownerName)
{
if (remoteServerStore.containsKey(ownerName))
{
return remoteServerStore.get(ownerName);
}
logger.debug("Getting remote server for {}", ownerName);
RemoteServer remoteServer = Factory.getInstance().createRemoteServer(new Scope(ownerName));
try
{
remoteServer.activate();
}
catch (InitializeException e)
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
remoteServerStore.put(ownerName, remoteServer);
return remoteServer;
}
// def _create_category_listener_if_needed(self, iu_category):
// '''Return (or create, store and return) a category listener.'''
// if iu_category in self._listener_store: return self._informer_store[iu_category]
......@@ -115,10 +250,25 @@ public class InputBuffer extends Buffer
{
return listenerStore.get(category);
}
Listener listener = Factory.getInstance().createListener(new Scope("/ipaaca/category/" + category));
Listener listener;
try
{
listener = Factory.getInstance().createListener(new Scope("/ipaaca/channel/" + this.channel + "/category/" + category));
}
catch (InitializeException e1)
{
throw new RuntimeException(e1);
}
listenerStore.put(category, listener);
categoryInterests.add(category);
listener.addHandler(new InputHandler(), true);
try
{
listener.addHandler(new InputHandler(), true);
}
catch (InterruptedException e1)
{
Thread.currentThread().interrupt();
}
logger.info("Added category listener for {}", category);
try
{
......@@ -128,6 +278,10 @@ public class InputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
return listener;
}
......@@ -141,7 +295,7 @@ public class InputBuffer extends Buffer
}
}
// def _handle_iu_events(self, event):
// '''Dispatch incoming IU events.
//
......@@ -185,11 +339,18 @@ public class InputBuffer extends Buffer
*/
private void handleIUEvents(Event event)
{
if(event.getData() instanceof RemoteMessageIU)
if (event.getData() instanceof RemoteMessageIU)
{
RemoteMessageIU rm = (RemoteMessageIU) event.getData();
if (messageStore.containsKey(rm.getUid())) {
logger.warn("Spurious RemoteMessage event: already got this UID: "+rm.getUid());
return;
}
//logger.info("Adding Message "+rm.getUid());
messageStore.put(rm.getUid(), rm);
callIuEventHandlers(rm.getUid(),false, IUEventType.ADDED, rm.getCategory());
//logger.info("Calling handlers for Message "+rm.getUid());
callIuEventHandlers(rm.getUid(),false, IUEventType.MESSAGE, rm.getCategory());
//logger.info("Removing Message "+rm.getUid());
messageStore.remove(rm.getUid());
}
else if (event.getData() instanceof RemotePushIU)
......@@ -220,7 +381,12 @@ public class InputBuffer extends Buffer
}
if (!iuStore.containsKey(iuLinkUpdate.getUid()))
{
logger.warn("Link update message for IU which we did not fully receive before.");
if (resendActive)
{
triggerResendRequest(event.getData(), getUniqueShortName());
} else {
logger.warn("Link update message for IU which we did not fully receive before.");
}
return;
}
RemotePushIU iu = this.iuStore.get(iuLinkUpdate.getUid());
......@@ -238,7 +404,12 @@ public class InputBuffer extends Buffer
}
if (!iuStore.containsKey(iuUpdate.getUid()))
{
logger.warn("Update message for IU which we did not fully receive before.");
if (resendActive)
{
triggerResendRequest(event.getData(), getUniqueShortName());
} else {
logger.warn("Update message for IU which we did not fully receive before.");
}
return;
}
RemotePushIU iu = this.iuStore.get(iuUpdate.getUid());
......@@ -258,7 +429,12 @@ public class InputBuffer extends Buffer
}
if (!iuStore.containsKey(iuc.getUid()))
{
logger.warn("Update message for IU which we did not fully receive before.");
if (resendActive)
{
triggerResendRequest(event.getData(), getUniqueShortName());
} else {
logger.warn("Update message for IU which we did not fully receive before.");
}
return;
}
RemotePushIU iu = this.iuStore.get(iuc.getUid());
......@@ -266,6 +442,72 @@ public class InputBuffer extends Buffer
iu.setRevision(iuc.getRevision());
callIuEventHandlers(iuc.getUid(), false, IUEventType.COMMITTED, iu.getCategory());
}
if (event.getData() instanceof IURetraction)
{
IURetraction iuc = (IURetraction) event.getData();
logger.debug("handleIUEvents invoked with an IURetraction: {}", iuc);
logger.debug("{}", this.getUniqueName());
if (!iuStore.containsKey(iuc.getUid()))
{
logger.warn("Update message for IU which we did not fully receive before.");
}
RemotePushIU iu = this.iuStore.get(iuc.getUid());
if (iu != null) {
iu.applyRetraction();
callIuEventHandlers(iuc.getUid(), false, IUEventType.RETRACTED, iu.getCategory());
}
}
}
}
private void triggerResendRequest(Object aiuObj, String hiddenScopeName)
{
String uid = null;
String writerName = null;
if (aiuObj instanceof IULinkUpdate) {
IULinkUpdate tmp = (IULinkUpdate)aiuObj;
uid = tmp.getUid();
writerName = tmp.getWriterName();
} else if (aiuObj instanceof IUPayloadUpdate) {
IUPayloadUpdate tmp = (IUPayloadUpdate)aiuObj;
uid = tmp.getUid();
writerName = tmp.getWriterName();
} else if (aiuObj instanceof IUCommission) {
IUCommission tmp = (IUCommission)aiuObj;
uid = tmp.getUid();
writerName = tmp.getWriterName();
}
RemoteServer rServer = null;
if (writerName != null)
rServer = getRemoteServer(writerName);
if ((rServer != null)&&(uid != null)) {
IUResendRequest iurr = IUResendRequest.newBuilder().setUid(uid).setHiddenScopeName(hiddenScopeName).build();
long rRevision = 0;
try
{
rRevision = (Long) rServer.call("resendRequest", iurr);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (rRevision == 0)
{
//throw new IUResendFailedException(aiu); // TODO
}
}
}
......@@ -277,7 +519,7 @@ public class InputBuffer extends Buffer
@Override
public AbstractIU getIU(String iuid)
{
if(iuStore.get(iuid)!=null)
if (iuStore.get(iuid) != null)
{
return iuStore.get(iuid);
}
......@@ -287,6 +529,14 @@ public class InputBuffer extends Buffer
}
}
public void addCategoryInterest(String... categories)
{
for(String cat:categories)
{
createCategoryListenerIfNeeded(cat);
}
}
public Collection<RemotePushIU> getIUs()
{
return iuStore.values();
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IntMessage;
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate.Builder;
import ipaaca.protobuf.Ipaaca.LinkSet;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
......@@ -16,6 +51,11 @@ import com.google.common.collect.SetMultimap;
public class LocalIU extends AbstractIU
{
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.PUSH;
}
private OutputBuffer outputBuffer;
......@@ -38,6 +78,15 @@ public class LocalIU extends AbstractIU
payload = new Payload(this);
}
public LocalIU(String category)
{
super(UUID.randomUUID().toString());
this.category = category;
revision = 1;
payload = new Payload(this);
}
// def _set_buffer(self, buffer):
// if self._buffer is not None:
// raise Exception('The IU is already in a buffer, cannot move it.')
......@@ -45,7 +94,7 @@ public class LocalIU extends AbstractIU
// self.owner_name = buffer.unique_name
// self._payload.owner_name = buffer.unique_name
//
public void setBuffer(OutputBuffer buffer)
protected void setBuffer(OutputBuffer buffer)
{
if (outputBuffer != null)
{
......@@ -62,6 +111,10 @@ public class LocalIU extends AbstractIU
synchronized (revisionLock)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (committed)
{
throw new IUCommittedException(this);
......@@ -78,6 +131,22 @@ public class LocalIU extends AbstractIU
}
}
private void internalRetract()
{
synchronized (revisionLock)
{
if (isRetracted())
return;
increaseRevisionNumber();
retracted = true;
if (outputBuffer != null)
{
outputBuffer.sendIURetraction(this);
}
}
}
private void increaseRevisionNumber()
{
revision++;
......@@ -105,6 +174,10 @@ public class LocalIU extends AbstractIU
@Override
void modifyLinks(boolean isDelta, SetMultimap<String, String> linksToAdd, SetMultimap<String, String> linksToRemove, String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isCommitted())
{
throw new IUCommittedException(this);
......@@ -138,7 +211,7 @@ public class LocalIU extends AbstractIU
removeSet.add(LinkSet.newBuilder().setType(entry.getKey()).addAllTargets(entry.getValue()).build());
}
outputBuffer.sendIULinkUpdate(this,
IULinkUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setWriterName(wName).setIsDelta(isDelta)
IULinkUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setWriterName(wName).setIsDelta(isDelta)
.addAllNewLinks(addSet).addAllLinksToRemove(removeSet).build());
}
}
......@@ -183,31 +256,81 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
// send update to remote holders
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("") // TODO: fix this, default in .proto?
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("STR")
.build();
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer).addNewItems(newItem).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
}
}
@Override
void putIntoPayload(Map<? extends String, ? extends String> newItems, String writer)
{
synchronized (getRevisionLock())
{
// set item locally
if (isCommitted())
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer);
for (Map.Entry<? extends String, ? extends String> item : newItems.entrySet())
{
PayloadItem newItem = PayloadItem.newBuilder().setKey(item.getKey()).setValue(item.getValue()).setType("STR")
.build();
builder.addNewItems(newItem);
}
IUPayloadUpdate update = builder.build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
}
}
@Override
public void commit()
{
if (isRetracted())
{
throw new IURetractedException(this);
}
internalCommit(null);
}
@Override
public void commit(String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
internalCommit(writerName);
}
@Override
public void retract()
{
internalRetract();
}
@Override
void removeFromPayload(Object key, String writer)
{
......@@ -217,11 +340,15 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
// send update to remote holders
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer).addKeysToRemove((String) key).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
......@@ -234,7 +361,7 @@ public class LocalIU extends AbstractIU
{
if (isPublished())
{
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(false)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(false)
.setWriterName(writerName == null ? getOwnerName() : writerName).addAllNewItems(newPayload).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
/**
* Local IU of Message sub-type. Can be handled like a normal IU, but on the remote side it is only existent during the handler calls.
* @author hvanwelbergen
......@@ -7,4 +41,19 @@ package ipaaca;
public class LocalMessageIU extends LocalIU
{
public LocalMessageIU()
{
super();
}
public LocalMessageIU(String category)
{
super(category);
}
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.MESSAGE;
}
}
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IURetraction;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.LinkSet;
......@@ -10,6 +45,8 @@ import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -18,7 +55,9 @@ import rsb.Informer;
import rsb.InitializeException;
import rsb.RSBException;
import rsb.patterns.DataCallback;
import rsb.patterns.EventCallback;
import rsb.patterns.LocalServer;
import rsb.Event;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
......@@ -27,6 +66,7 @@ import com.google.common.collect.SetMultimap;
* An OutputBuffer that holds local IUs.
* @author hvanwelbergen
*/
@Slf4j
public class OutputBuffer extends Buffer
{
......@@ -34,6 +74,7 @@ public class OutputBuffer extends Buffer
private Map<String, Informer<Object>> informerStore = new HashMap<String, Informer<Object>>(); // category -> informer map
private final static Logger logger = LoggerFactory.getLogger(OutputBuffer.class.getName());
private IUStore<LocalIU> iuStore = new IUStore<LocalIU>();
private String channel = "default";
// def __init__(self, owning_component_name, participant_config=None):
// '''Create an Output Buffer.
......@@ -55,10 +96,20 @@ public class OutputBuffer extends Buffer
* @param owningComponentName name of the entity that own this buffer
*/
public OutputBuffer(String owningComponentName)
{
this(owningComponentName, "default");
}
/**
* @param owningComponentName name of the entity that own this buffer
* @param channel name of the ipaaca channel this buffer is using
*/
public OutputBuffer(String owningComponentName, String ipaaca_channel)
{
super(owningComponentName);
uniqueName = "/ipaaca/component/" + owningComponentName + "ID" + uuid + "/OB";
uniqueName = "/ipaaca/component/" + getUniqueShortName() + "/OB";
logger.debug("Creating server for {}", uniqueName);
server = Factory.getInstance().createLocalServer(uniqueName);
try
......@@ -66,46 +117,74 @@ public class OutputBuffer extends Buffer
server.addMethod("updatePayload", new RemoteUpdatePayload());
server.addMethod("updateLinks", new RemoteUpdateLinks());
server.addMethod("commit", new RemoteCommit());
// add method to trigger a resend request. (dlw)
server.addMethod("resendRequest", new RemoteResendRequest());
server.activate();
}
catch (InitializeException e)
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
this.channel = ipaaca_channel;
}
private final class RemoteUpdatePayload implements DataCallback<Integer, IUPayloadUpdate>
private final class RemoteUpdatePayload extends EventCallback //DataCallback<Long, IUPayloadUpdate>
{
@Override
public Integer invoke(IUPayloadUpdate data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteUpdate");
return remoteUpdatePayload(data);
long result = remoteUpdatePayload((IUPayloadUpdate) request.getData());
//System.out.println("remoteUpdatePayload yielded revision "+result);
return new Event(Long.class, new Long(result));
}
}
/*private final class RemoteUpdatePayload extends DataCallback<Long, IUPayloadUpdate>
{
@Override
public Long invoke(IUPayloadUpdate data) throws Throwable
{
logger.debug("remoteUpdate");
return remoteUpdatePayload(data);
}
}*/
private final class RemoteUpdateLinks implements DataCallback<Integer, IULinkUpdate>
private final class RemoteUpdateLinks extends EventCallback // DataCallback<Long, IULinkUpdate>
{
@Override
public Integer invoke(IULinkUpdate data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteUpdateLinks");
return remoteUpdateLinks(data);
return new Event(Long.class, new Long(remoteUpdateLinks((IULinkUpdate) request.getData())));
}
}
private final class RemoteCommit implements DataCallback<Integer, IUCommission>
private final class RemoteCommit extends EventCallback //DataCallback<Long, IUCommission>
{
@Override
public Integer invoke(IUCommission data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteCommit");
return remoteCommit(data);
return new Event(Long.class, new Long(remoteCommit((IUCommission) request.getData())));
}
}
private final class RemoteResendRequest extends EventCallback //DataCallback<Long, IUResendRequest>
{
@Override
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteResendRequest");
return new Event(Long.class, new Long(remoteResendRequest((IUResendRequest) request.getData())));
}
}
// def _remote_update_payload(self, update):
......@@ -132,7 +211,7 @@ public class OutputBuffer extends Buffer
* Apply a remotely requested update to one of the stored IUs.
* @return 0 if not updated, IU version number otherwise
*/
int remoteUpdatePayload(IUPayloadUpdate update)
long remoteUpdatePayload(IUPayloadUpdate update)
{
if (!iuStore.containsKey(update.getUid()))
{
......@@ -153,14 +232,20 @@ public class OutputBuffer extends Buffer
{
iu.getPayload().remove(k, update.getWriterName());
}
for (PayloadItem pli : update.getNewItemsList())
if (update.getNewItemsList().size() > 0)
{
iu.getPayload().put(pli.getKey(), pli.getValue(), update.getWriterName());
HashMap<String, String> payloadUpdate = new HashMap<String, String>();
for (PayloadItem pli : update.getNewItemsList())
{
payloadUpdate.put(pli.getKey(), pli.getValue());
// //iu.getPayload().put(pli.getKey(), pli.getValue(), update.getWriterName());
}
iu.getPayload().putAll(payloadUpdate, update.getWriterName());
}
}
else
{
iu.setPayload(update.getNewItemsList(), update.getWriterName());
}
callIuEventHandlers(update.getUid(), true, IUEventType.UPDATED, iu.getCategory());
......@@ -171,7 +256,7 @@ public class OutputBuffer extends Buffer
* Apply a remotely requested update to one of the stored IUs.
* @return 0 if not updated, IU version number otherwise
*/
int remoteUpdateLinks(IULinkUpdate update)
long remoteUpdateLinks(IULinkUpdate update)
{
if (!iuStore.containsKey(update.getUid()))
{
......@@ -234,7 +319,7 @@ public class OutputBuffer extends Buffer
/**
* Apply a remotely requested commit to one of the stored IUs.
*/
private int remoteCommit(IUCommission iuc)
private long remoteCommit(IUCommission iuc)
{
if (!iuStore.containsKey(iuc.getUid()))
{
......@@ -260,6 +345,35 @@ public class OutputBuffer extends Buffer
}
}
/*
* Resend an requested iu over the specific hidden channel. (dlw) TODO
*/
private long remoteResendRequest(IUResendRequest iu_resend_request_pack)
{
if (!iuStore.containsKey(iu_resend_request_pack.getUid()))
{
logger.warn("Remote InBuffer tried to spuriously write non-existent IU {}", iu_resend_request_pack.getUid());
return 0;
}
AbstractIU iu = iuStore.get(iu_resend_request_pack.getUid());
if ((iu_resend_request_pack.hasHiddenScopeName() == true)&&(!iu_resend_request_pack.getHiddenScopeName().equals("")))
{
Informer<Object> informer = getInformer(iu_resend_request_pack.getHiddenScopeName());
try
{
informer.publish(iu);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
return iu.getRevision();
} else
{
return 0;
}
}
// def _get_informer(self, iu_category):
// '''Return (or create, store and return) an informer object for IUs of the specified category.'''
// if iu_category in self._informer_store:
......@@ -281,12 +395,19 @@ public class OutputBuffer extends Buffer
{
return informerStore.get(category);
}
Informer<Object> informer = Factory.getInstance().createInformer("/ipaaca/category/" + category);
Informer<Object> informer;
try
{
informer = Factory.getInstance().createInformer("/ipaaca/channel/" + this.channel + "/category/" + category);
}
catch (InitializeException e1)
{
throw new RuntimeException(e1);
}
informerStore.put(category, informer);
logger.info("Added informer on " + category);
logger.info("Added informer on channel " + this.channel + " and category " + category);
// XXX new in java version, apperently informers need activation and deactivation
try
{
informer.activate();
......@@ -295,6 +416,10 @@ public class OutputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
return informer;
}
......@@ -315,7 +440,10 @@ public class OutputBuffer extends Buffer
{
throw new IUPublishedException(iu);
}
iuStore.put(iu.getUid(), iu);
if(!(iu instanceof LocalMessageIU))
{
iuStore.put(iu.getUid(), iu);
}
iu.setBuffer(this);
publishIU(iu);
}
......@@ -324,12 +452,12 @@ public class OutputBuffer extends Buffer
// '''Publish an IU.'''
// informer = self._get_informer(iu._category)
// informer.publishData(iu)
public void publishIU(AbstractIU iu)
private void publishIU(AbstractIU iu)
{
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(iu);
informer.publish(iu);
}
catch (RSBException e)
{
......@@ -361,14 +489,28 @@ public class OutputBuffer extends Buffer
* to enable remote components to filter out updates that originated
* from their own operations
*/
public void sendIUCommission(AbstractIU iu, String writerName)
protected void sendIUCommission(AbstractIU iu, String writerName)
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(iu.getUid()).setRevision((int) iu.getRevision())
.setWriterName(writerName == null ? iu.getOwnerName() : writerName).build();
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.publish(iuc);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
}
protected void sendIURetraction(AbstractIU iu)
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(iu.getUid()).setRevision(iu.getRevision())
.setWriterName(iu.getOwnerName() != null ? iu.getOwnerName() : writerName).build();
IURetraction iuc = Ipaaca.IURetraction.newBuilder().setUid(iu.getUid()).setRevision((int) iu.getRevision()).build();
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(iuc);
informer.publish(iuc);
}
catch (RSBException e)
{
......@@ -403,12 +545,12 @@ public class OutputBuffer extends Buffer
// informer = self._get_informer(iu._category)
// informer.publishData(payload_update)
public void sendIUPayloadUpdate(AbstractIU iu, IUPayloadUpdate update)
protected void sendIUPayloadUpdate(AbstractIU iu, IUPayloadUpdate update)
{
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......@@ -416,12 +558,12 @@ public class OutputBuffer extends Buffer
}
}
public void sendIULinkUpdate(AbstractIU iu, IULinkUpdate update)
protected void sendIULinkUpdate(AbstractIU iu, IULinkUpdate update)
{
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......@@ -437,10 +579,32 @@ public class OutputBuffer extends Buffer
public void close()
{
server.deactivate();
try
{
server.deactivate();
}
catch (RSBException e)
{
log.warn("RSBException on deactivating server in close", e);
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
for (Informer<?> informer : informerStore.values())
{
informer.deactivate();
try
{
informer.deactivate();
}
catch (RSBException e)
{
log.warn("RSBException on deactivating informer {} in close", e, informer.toString());
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
}
}
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import org.apache.commons.lang.StringEscapeUtils;
import com.google.common.collect.ImmutableSet;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
......@@ -15,7 +52,7 @@ import java.util.Set;
*/
public class Payload implements Map<String, String>
{
private Map<String, String> map = new HashMap<String, String>();
private Map<String, String> map = Collections.synchronizedMap(new HashMap<String, String>());
private final AbstractIU iu;
public Payload(AbstractIU iu)
......@@ -23,18 +60,6 @@ public class Payload implements Map<String, String>
this.iu = iu;
}
// def __init__(self, remote_push_iu, new_payload):
// """Create remote payload object.
//
// Keyword arguments:
// remote_push_iu -- remote IU holding this payload
// new_payload -- payload dict to initialise this remote payload with
// """
// super(RemotePushPayload, self).__init__()
// self._remote_push_iu = remote_push_iu
// if new_payload is not None:
// for k,v in new_payload.items():
// dict.__setitem__(self, k, v)
public Payload(AbstractIU iu, List<PayloadItem> payloadItems)
{
this(iu, payloadItems, null);
......@@ -60,31 +85,45 @@ public class Payload implements Map<String, String>
public void set(Map<String, String> newPayload, String writerName)
{
iu.setPayload(newPayload, writerName);
map.clear();
map.putAll(newPayload);
synchronized(map)
{
map.clear();
map.putAll(newPayload);
}
}
public void set(List<PayloadItem> newPayload, String writerName)
{
iu.handlePayloadSetting(newPayload, writerName);
map.clear();
for (PayloadItem item : newPayload)
synchronized(map)
{
map.put(item.getKey(), item.getValue());
map.clear();
for (PayloadItem item : newPayload)
{
map.put(item.getKey(), pseudoConvertFromJSON(item.getValue(), item.getType()));
}
}
}
public String pseudoConvertFromJSON(String value, String type) {
if (type.equals("JSON")) {
if (value.startsWith("\"")) {
//return value.replaceAll("\\\"", "");
return StringEscapeUtils.unescapeJava(value.substring(1, value.length() - 1));
} else if (value.startsWith("{") || value.startsWith("[") || value.matches("true") || value.matches("false") || value.matches("-?[0-9]*[.,]?[0-9][0-9]*.*")) {
return value;
} else if (value.equals("null")) {
return "";
}
}
return value;
}
// def _remotely_enforced_setitem(self, k, v):
// """Sets an item when requested remotely."""
// return dict.__setitem__(self, k, v)
void enforcedSetItem(String key, String value)
{
map.put(key, value);
}
// def _remotely_enforced_delitem(self, k):
// """Deletes an item when requested remotely."""
// return dict.__delitem__(self, k)
void enforcedRemoveItem(String key)
{
map.remove(key);
......@@ -106,9 +145,12 @@ public class Payload implements Map<String, String>
return map.containsValue(value);
}
public Set<java.util.Map.Entry<String, String>> entrySet()
/**
* Provides an immutable copy of the entryset of the Payload
*/
public ImmutableSet<java.util.Map.Entry<String, String>> entrySet()
{
return map.entrySet();
return ImmutableSet.copyOf(map.entrySet());
}
public boolean equals(Object o)
......@@ -136,31 +178,6 @@ public class Payload implements Map<String, String>
return map.keySet();
}
// def __setitem__(self, k, v):
// """Set item in this payload.
//
// Requests item setting from the OutputBuffer holding the local version
// of this IU. Returns when permission is granted and item is set;
// otherwise raises an IUUpdateFailedError.
// """
// if self._remote_push_iu.committed:
// raise IUCommittedError(self._remote_push_iu)
// if self._remote_push_iu.read_only:
// raise IUReadOnlyError(self._remote_push_iu)
// requested_update = IUPayloadUpdate(
// uid=self._remote_push_iu.uid,
// revision=self._remote_push_iu.revision,
// is_delta=True,
// writer_name=self._remote_push_iu.buffer.unique_name,
// new_items={k:v},
// keys_to_remove=[])
// remote_server = self._remote_push_iu.buffer._get_remote_server(self._remote_push_iu)
// new_revision = remote_server.updatePayload(requested_update)
// if new_revision == 0:
// raise IUUpdateFailedError(self._remote_push_iu)
// else:
// self._remote_push_iu._revision = new_revision
// dict.__setitem__(self, k, v)
/**
* Set item in this payload.
* Requests item setting from the OutputBuffer holding the local version
......@@ -173,32 +190,6 @@ public class Payload implements Map<String, String>
return map.put(key, value);
}
//
// def __delitem__(self, k):
// """Delete item in this payload.
//
// Requests item deletion from the OutputBuffer holding the local version
// of this IU. Returns when permission is granted and item is deleted;
// otherwise raises an IUUpdateFailedError.
// """
// if self._remote_push_iu.committed:
// raise IUCommittedError(self._remote_push_iu)
// if self._remote_push_iu.read_only:
// raise IUReadOnlyError(self._remote_push_iu)
// requested_update = IUPayloadUpdate(
// uid=self._remote_push_iu.uid,
// revision=self._remote_push_iu.revision,
// is_delta=True,
// writer_name=self._remote_push_iu.buffer.unique_name,
// new_items={},
// keys_to_remove=[k])
// remote_server = self._remote_push_iu.buffer._get_remote_server(self._remote_push_iu)
// new_revision = remote_server.updatePayload(requested_update)
// if new_revision == 0:
// raise IUUpdateFailedError(self._remote_push_iu)
// else:
// self._remote_push_iu._revision = new_revision
// dict.__delitem__(self, k)
/**
* Delete item in this payload.//
* Requests item deletion from the OutputBuffer holding the local version
......@@ -216,9 +207,20 @@ public class Payload implements Map<String, String>
return put(key, value, null);
}
public void putAll(Map<? extends String, ? extends String> m)
public void putAll(Map<? extends String, ? extends String> newItems)
{
throw new RuntimeException("Not implemented");
putAll(newItems, null);
}
public void putAll(Map<? extends String, ? extends String> newItems, String writer)
{
iu.putIntoPayload(newItems, writer);
map.putAll(newItems);
}
public void merge(Map<? extends String, ? extends String> items) {
putAll(items, null);
}
public String remove(Object key)
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
......@@ -12,6 +46,11 @@ import com.google.common.collect.SetMultimap;
public class RemoteMessageIU extends AbstractIU
{
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.MESSAGE;
}
public RemoteMessageIU(String uid)
{
super(uid);
......@@ -25,6 +64,12 @@ public class RemoteMessageIU extends AbstractIU
committed = true;
}
@Override
public void retract()
{
log.info("Retracting a RemoteMessage has no effect.");
}
@Override
public void commit(String writerName)
{
......@@ -48,6 +93,15 @@ public class RemoteMessageIU extends AbstractIU
payload.put(key,value);
log.info("Info: modifying a RemoteMessage only has local effects");
}
void putIntoPayload(Map<? extends String, ? extends String> newItems, String writer) {
for (Map.Entry<? extends String, ? extends String> item : newItems.entrySet())
{
payload.put(item.getKey(), item.getValue());
//System.out.println(entry.getKey() + "/" + entry.getValue());
}
log.info("Info: modifying a RemoteMessage only has local effects");
}
@Override
void removeFromPayload(Object key, String writer)
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca;
import ipaaca.protobuf.Ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.LinkSet;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate.Builder;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -30,19 +67,16 @@ public class RemotePushIU extends AbstractIU
private final static Logger logger = LoggerFactory.getLogger(RemotePushIU.class.getName());
private InputBuffer inputBuffer;
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.PUSH;
}
public InputBuffer getInputBuffer()
{
return inputBuffer;
}
// def __init__(self, uid, revision, read_only, owner_name, category, type, committed, payload):
// super(RemotePushIU, self).__init__(uid=uid, access_mode=IUAccessMode.PUSH, read_only=read_only)
// self._revision = revision
// self._category = category
// self.owner_name = owner_name
// self._type = type
// self._committed = committed
// self._payload = RemotePushPayload(remote_push_iu=self, new_payload=payload)
public RemotePushIU(String uid)
{
super(uid);
......@@ -56,9 +90,9 @@ public class RemotePushIU extends AbstractIU
}
@Override
public void commit()
public void retract()
{
commit(null);
logger.info("Retracting a RemoteIU has no effect.");
}
void putIntoPayload(String key, String value, String writer)
......@@ -67,25 +101,43 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("").build();// TODO use default type in .proto
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision(getRevision())
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("STR").build();
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).addNewItems(newItem).build();
RemoteServer server = getInputBuffer().getRemoteServer(this);
logger.debug("Remote server has methods {}", server.getMethods());
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
//System.out.println("calling remote updatePayload ...");
newRevision = (Long) server.call("updatePayload", update);
//System.out.println(" ... done");
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -93,28 +145,77 @@ public class RemotePushIU extends AbstractIU
setRevision(newRevision);
}
// def commit(self):
// """Commit to this IU."""
// if self.read_only:
// raise IUReadOnlyError(self)
// if self._committed:
// # ignore commit requests when already committed
// return
// else:
// commission_request = iuProtoBuf_pb2.IUCommission()
// commission_request.uid = self.uid
// commission_request.revision = self.revision
// commission_request.writer_name = self.buffer.unique_name
// remote_server = self.buffer._get_remote_server(self)
// new_revision = remote_server.commit(commission_request)
// if new_revision == 0:
// raise IUUpdateFailedError(self)
// else:
// self._revision = new_revision
// self._committed = True
@Override
void putIntoPayload(Map<? extends String, ? extends String> newItems, String writer)
{
if (isCommitted())
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(getBuffer().getUniqueName());
for (Map.Entry<? extends String, ? extends String> item : newItems.entrySet())
{
PayloadItem newItem = PayloadItem.newBuilder().setKey(item.getKey()).setValue(item.getValue()).setType("STR")
.build();
builder.addNewItems(newItem);
}
IUPayloadUpdate update = builder.build();
RemoteServer server = getInputBuffer().getRemoteServer(this);
logger.debug("Remote server has methods {}", server.getMethods());
long newRevision;
try
{
newRevision = (Long) server.call("updatePayload", update);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
}
System.err.print("************************ ");
System.err.println(newRevision);
setRevision(newRevision);
}
@Override
public void commit()
{
commit(null);
}
@Override
public void commit(String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
......@@ -125,18 +226,30 @@ public class RemotePushIU extends AbstractIU
}
else
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(getUid()).setRevision(getRevision())
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).build();
RemoteServer server = inputBuffer.getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("commit", iuc);
newRevision = (Long) server.call("commit", iuc);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUCommittedException(this);
......@@ -149,17 +262,6 @@ public class RemotePushIU extends AbstractIU
}
}
// def __str__(self):
// s = "RemotePushIU{ "
// s += "uid="+self._uid+" "
// s += "(buffer="+(self.buffer.unique_name if self.buffer is not None else "<None>")+") "
// s += "owner_name=" + ("<None>" if self.owner_name is None else self.owner_name) + " "
// s += "payload={ "
// for k,v in self.payload.items():
// s += k+":'"+v+"', "
// s += "} "
// s += "}"
// return s
@Override
public String toString()
{
......@@ -178,37 +280,11 @@ public class RemotePushIU extends AbstractIU
return b.toString();
}
//
// def _get_payload(self):
// return self._payload
public Payload getPayload()
{
return payload;
}
// def _set_payload(self, new_pl):
// if self.committed:
// raise IUCommittedError(self)
// if self.read_only:
// raise IUReadOnlyError(self)
// requested_update = IUPayloadUpdate(
// uid=self.uid,
// revision=self.revision,
// is_delta=False,
// writer_name=self.buffer.unique_name,
// new_items=new_pl,
// keys_to_remove=[])
// remote_server = self.buffer._get_remote_server(self)
// new_revision = remote_server.updatePayload(requested_update)
// if new_revision == 0:
// raise IUUpdateFailedError(self)
// else:
// self._revision = new_revision
// self._payload = RemotePushPayload(remote_push_iu=self, new_payload=new_pl)
// payload = property(
// fget=_get_payload,
// fset=_set_payload,
// doc='Payload dictionary of the IU.')
@Override
public void setPayload(List<PayloadItem> newItems, String writerName)
{
......@@ -216,23 +292,39 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
IUPayloadUpdate iuu = IUPayloadUpdate.newBuilder().setRevision(getRevision()).setIsDelta(false).setUid(getUid())
IUPayloadUpdate iuu = IUPayloadUpdate.newBuilder().setRevision((int) getRevision()).setIsDelta(false).setUid(getUid())
.addAllNewItems(newItems).setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").build();
RemoteServer server = inputBuffer.getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", iuu);
newRevision = (Long) server.call("updatePayload", iuu);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -244,37 +336,33 @@ public class RemotePushIU extends AbstractIU
}
}
// def _apply_update(self, update):
// """Apply a IUPayloadUpdate to the IU."""
// self._revision = update.revision
// if update.is_delta:
// for k in update.keys_to_remove: self.payload._remotely_enforced_delitem(k)
// for k, v in update.new_items.items(): self.payload._remotely_enforced_setitem(k, v)
// else:
// # using '_payload' to circumvent the local writing methods
// self._payload = RemotePushPayload(remote_push_iu=self, new_payload=update.new_items)
/**
* Apply a IUPayloadUpdate to the IU.
* @param update
*/
public void applyUpdate(IUPayloadUpdate update)
{
revision = update.getRevision();
if (update.getIsDelta())
{
for (String key : update.getKeysToRemoveList())
{
payload.enforcedRemoveItem(key);
}
for (PayloadItem item : update.getNewItemsList())
{
payload.enforcedSetItem(item.getKey(), item.getValue());
}
}
else
{
payload = new Payload(this, update.getNewItemsList());
}
public void applyUpdate(IUPayloadUpdate update) {
revision = update.getRevision();
if (update.getIsDelta()) {
for (String key : update.getKeysToRemoveList()) {
payload.enforcedRemoveItem(key);
}
for (PayloadItem item : update.getNewItemsList()) {
if (item.getType().equals("STR")) {
payload.enforcedSetItem(item.getKey(), item.getValue());
} else if (item.getType().equals("JSON")) {
String value = item.getValue();
if (value.startsWith("\"")) {
payload.enforcedSetItem(item.getKey(), value.replaceAll("\\\"", ""));
} else if (value.startsWith("{") || value.startsWith("[") || value.matches("true") || value.matches("false") || value.matches("-?[0-9]*[.,]?[0-9][0-9]*.*")) {
payload.enforcedSetItem(item.getKey(), value);
} else if (value.equals("null")) {
payload.enforcedSetItem(item.getKey(), "");
}
}
}
} else {
payload = new Payload(this, update.getNewItemsList());
}
}
public void applyLinkUpdate(IULinkUpdate update)
......@@ -310,14 +398,16 @@ public class RemotePushIU extends AbstractIU
}
// def _apply_commission(self):
// """Apply commission to the IU"""
// self._committed = True
public void applyCommmision()
{
committed = true;
}
public void applyRetraction()
{
retracted = true;
}
@Override
void removeFromPayload(Object key, String writer)
{
......@@ -325,22 +415,38 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision(getRevision())
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).addKeysToRemove((String) key).build();
RemoteServer server = getInputBuffer().getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
newRevision = (Long) server.call("updatePayload", update);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -348,25 +454,7 @@ public class RemotePushIU extends AbstractIU
setRevision(newRevision);
}
// def _modify_payload(self, payload, is_delta=True, new_items={}, keys_to_remove=[], writer_name=None):
// """Modify the payload: add or remove item from this payload remotely and send update."""
// if self.committed:
// raise IUCommittedError(self)
// if self.read_only:
// raise IUReadOnlyError(self)
// requested_update = IUPayloadUpdate(
// uid=self.uid,
// revision=self.revision,
// is_delta=is_delta,
// writer_name=self.buffer.unique_name,
// new_items=new_items,
// keys_to_remove=keys_to_remove)
// remote_server = self.buffer._get_remote_server(self)
// new_revision = remote_server.updatePayload(requested_update)
// if new_revision == 0:
// raise IUUpdateFailedError(self)
// else:
// self._revision = new_revision
@Override
void modifyLinks(boolean isDelta, SetMultimap<String, String> linksToAdd, SetMultimap<String, String> linksToRemove, String writerName)
{
......@@ -374,6 +462,10 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
......@@ -391,16 +483,28 @@ public class RemotePushIU extends AbstractIU
}
IULinkUpdate update = IULinkUpdate.newBuilder().addAllLinksToRemove(removeLinkSet).addAllNewLinks(newLinkSet).setIsDelta(isDelta)
.setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").setUid(getUid()).setRevision(getRevision()).build();
int newRevision;
.setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").setUid(getUid()).setRevision((int) getRevision()).build();
long newRevision;
try
{
newRevision = (Integer) server.call("updateLinks", update);
newRevision = (Long) server.call("updateLinks", update);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......
package ipaaca.util;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
import ipaaca.LocalIU;
import ipaaca.OutputBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableSet;
/**
* A simple key-value blackboard
* @author hvanwelbergen
*/
public class Blackboard
{
private final OutputBuffer outBuffer;
private final InputBuffer inBuffer;
private final LocalIU iu;
private final ComponentNotifier notifier;
private static final String DUMMY_KEY = "DUMMY_KEY";
public static final String MESSAGE_SUFFIX = "MESSAGE";
private int dummyValue = 0;
private List<BlackboardUpdateListener> listeners = Collections.synchronizedList(new ArrayList<BlackboardUpdateListener>());
public Blackboard(String id, String category)
{
this(id, category, "default");
}
private void updateListeners()
{
synchronized (listeners)
{
for (BlackboardUpdateListener listener : listeners)
{
listener.update();
}
}
}
public Blackboard(String id, String category, String channel)
{
outBuffer = new OutputBuffer(id, channel);
iu = new LocalIU(category);
outBuffer.add(iu);
outBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
updateListeners();
}
});
inBuffer = new InputBuffer(id, ImmutableSet.of(ComponentNotifier.NOTIFY_CATEGORY, category + MESSAGE_SUFFIX), channel);
notifier = new ComponentNotifier(id, category, ImmutableSet.of(category), new HashSet<String>(), outBuffer, inBuffer);
notifier.addNotificationHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iuNotify, IUEventType type, boolean local)
{
dummyValue++;
iu.getPayload().put(DUMMY_KEY, "" + dummyValue);
}
});
notifier.initialize();
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iuMessage, IUEventType type, boolean local)
{
iu.getPayload().putAll(iuMessage.getPayload());
updateListeners();
}
}, ImmutableSet.of(category + MESSAGE_SUFFIX));
}
public String put(String key, String value)
{
return iu.getPayload().put(key, value);
}
public void putAll(Map<String, String> newItems)
{
iu.getPayload().putAll(newItems);
}
/**
* Get the value corresponding to the key, or null if it is not available
*/
public String get(String key)
{
return iu.getPayload().get(key);
}
public void addUpdateListener(BlackboardUpdateListener listener)
{
listeners.add(listener);
}
public Set<String> keySet()
{
return iu.getPayload().keySet();
}
public Set<Map.Entry<String, String>> entrySet()
{
return iu.getPayload().entrySet();
}
public Collection<String> values()
{
return iu.getPayload().values();
}
public void close()
{
outBuffer.close();
inBuffer.close();
}
}
package ipaaca.util;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
import ipaaca.LocalMessageIU;
import ipaaca.OutputBuffer;
/**
* Client to get/set key value pairs on a Blackboard
* @author hvanwelbergen
*
*/
public class BlackboardClient
{
private final InputBuffer inBuffer;
private final OutputBuffer outBuffer;
private List<BlackboardUpdateListener> listeners = Collections.synchronizedList(new ArrayList<BlackboardUpdateListener>());
private final String category;
public BlackboardClient(String id, String category)
{
this(id, category, "default");
}
public BlackboardClient(String id, String category, String channel)
{
this.category = category;
inBuffer = new InputBuffer(id, ImmutableSet.of(category, ComponentNotifier.NOTIFY_CATEGORY), channel);
inBuffer.setResendActive(true);
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
synchronized (listeners)
{
for (BlackboardUpdateListener listener : listeners)
{
listener.update();
}
}
}
}, ImmutableSet.of(category));
outBuffer = new OutputBuffer(id);
ComponentNotifier notifier = new ComponentNotifier(id, category, new HashSet<String>(), ImmutableSet.of(category),
outBuffer, inBuffer);
notifier.initialize();
}
public void close()
{
inBuffer.close();
outBuffer.close();
}
public void waitForBlackboardConnection()
{
while(inBuffer.getIUs().isEmpty());
}
public String get(String key)
{
if (inBuffer.getIUs().isEmpty())
{
return null;
}
return inBuffer.getIUs().iterator().next().getPayload().get(key);
}
public void put(String key, String value)
{
LocalMessageIU iu = new LocalMessageIU(category+Blackboard.MESSAGE_SUFFIX);
iu.getPayload().put(key,value);
outBuffer.add(iu);
}
public void putAll(Map<String,String> values)
{
LocalMessageIU iu = new LocalMessageIU(category+Blackboard.MESSAGE_SUFFIX);
iu.getPayload().putAll(values);
outBuffer.add(iu);
}
private boolean hasIU()
{
return !inBuffer.getIUs().isEmpty();
}
private AbstractIU getIU()
{
return inBuffer.getIUs().iterator().next();
}
public Set<String> keySet()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().keySet();
}
public Set<Map.Entry<String, String>> entrySet()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().entrySet();
}
public Collection<String> values()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().values();
}
public void addUpdateListener(BlackboardUpdateListener listener)
{
listeners.add(listener);
}
}
package ipaaca.util;
public interface BlackboardUpdateListener
{
void update();
}
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca.util;
import ipaaca.AbstractIU;
......@@ -42,7 +74,7 @@ public class ComponentNotifier
private final InputBuffer inBuffer;
private List<HandlerFunctor> handlers = Collections.synchronizedList(new ArrayList<HandlerFunctor>());
private volatile boolean isInitialized = false;
private final BlockingQueue<String> receiverQueue = new LinkedBlockingQueue<>();
private final BlockingQueue<String> receiverQueue = new LinkedBlockingQueue<String>();
private class ComponentNotifyHandler implements HandlerFunctor
{
......@@ -79,7 +111,7 @@ public class ComponentNotifier
*/
public void waitForReceivers(ImmutableSet<String> categories)
{
Set<String> unhandledCategories = new HashSet<>(categories);
Set<String> unhandledCategories = new HashSet<String>(categories);
while(!unhandledCategories.isEmpty())
{
try
......@@ -129,7 +161,7 @@ public class ComponentNotifier
{
if(!isInitialized)
{
inBuffer.registerHandler(new IUEventHandler(new ComponentNotifyHandler(), EnumSet.of(IUEventType.ADDED), ImmutableSet
inBuffer.registerHandler(new IUEventHandler(new ComponentNotifyHandler(), EnumSet.of(IUEventType.ADDED, IUEventType.MESSAGE), ImmutableSet
.of(NOTIFY_CATEGORY)));
submitNotify(true);
isInitialized = true;
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2016 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca.util;
import ipaaca.LocalMessageIU;
import ipaaca.OutputBuffer;
import java.util.HashMap;
import java.util.UUID;
import org.apache.commons.lang.StringUtils;
public class IpaacaLogger {
private static OutputBuffer ob;
private static final Object lock = new Object();
private static boolean SEND_IPAACA_LOGS = true;
private static String MODULE_NAME = "???";
private static void initializeOutBuffer() {
synchronized (lock) {
if (ob == null) {
ob = new OutputBuffer("LogSender");
}
}
}
public static void setModuleName(String name) {
synchronized (lock) {
MODULE_NAME = name;
}
}
public static void setLogFileName(String fileName, String logMode) {
initializeOutBuffer();
LocalMessageIU msg = new LocalMessageIU("log");
HashMap<String, String> pl = new HashMap<String, String>();
pl.put("cmd", "open_log_file");
pl.put("filename", fileName);
if (logMode != null) {
if (logMode.equals("append") ||
logMode.equals("overwrite") ||
logMode.equals("timestamp")) {
pl.put("existing", logMode);
} else {
return;
}
}
ob.add(msg);
}
public static void sendIpaacaLogs(boolean flag) {
synchronized (lock) {
SEND_IPAACA_LOGS = flag;
}
}
private static void logConsole(String level, String text, float now, String function, String thread) {
for(String line: text.split("\n")) {
System.out.println("[" + level + "] " + thread + " " + function + " " + line);
function = StringUtils.leftPad("", function.length(), ' ');
thread = StringUtils.leftPad("", thread.length(), ' ');
}
}
private static void logIpaaca(String level, String text, float now, String function, String thread) {
initializeOutBuffer();
LocalMessageIU msg = new LocalMessageIU("log");
HashMap<String, String> pl = new HashMap<String, String>();
pl.put("module", MODULE_NAME);
pl.put("function", function);
pl.put("level", level);
pl.put("time", String.format("%.3f", now));
pl.put("thread", thread);
pl.put("uuid", UUID.randomUUID().toString());
pl.put("text", text);
msg.setPayload(pl);
ob.add(msg);
}
private static String getCallerName() {
String function = Thread.currentThread().getStackTrace()[3].getClassName();
function += "." + Thread.currentThread().getStackTrace()[3].getMethodName();
return function;
}
public static void logError(String msg) {
logError(msg, System.currentTimeMillis(), getCallerName());
}
public static void logError(String msg, float now) {
logError(msg, now, getCallerName());
}
private static void logError(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("ERROR", msg, now, callerName, thread);
}
logConsole("ERROR", msg, now, callerName, thread);
}
public static void logWarn(String msg) {
logWarn(msg, System.currentTimeMillis(), getCallerName());
}
public static void logWarn(String msg, float now) {
logWarn(msg, now, getCallerName());
}
private static void logWarn(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("WARN", msg, now, callerName, thread);
}
logConsole("WARN", msg, now, callerName, thread);
}
public static void logInfo(String msg) {
logInfo(msg, System.currentTimeMillis(), getCallerName());
}
public static void logInfo(String msg, float now) {
logInfo(msg, now, getCallerName());
}
private static void logInfo(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("INFO", msg, now, callerName, thread);
}
logConsole("INFO", msg, now, callerName, thread);
}
public static void logDebug(String msg) {
logDebug(msg, System.currentTimeMillis(), getCallerName());
}
public static void logDebug(String msg, float now) {
logDebug(msg, now, getCallerName());
}
private static void logDebug(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("DEBUG", msg, now, callerName, thread);
}
logConsole("DEBUG", msg, now, callerName, thread);
}
}
package ipaaca.util.communication;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventHandler;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
/**
* Obtain a IU in the future. Usage:<br>
* FutureIU fu = FutureIU("componentx", "status", "started"); //wait for componentx to send a message that is it fully started<br>
* [Start componentx, assumes that component x will send a message or other iu with status=started in the payload]<br>
* AbstractIU iu = fu.take(); //get the actual IU
* @author hvanwelbergen
*/
public class FutureIU
{
private final InputBuffer inBuffer;
private final BlockingQueue<AbstractIU> queue = new ArrayBlockingQueue<AbstractIU>(1);
private final IUEventHandler handler;
public FutureIU(String category, final String idKey, final String idVal, InputBuffer inBuffer)
{
this.inBuffer = inBuffer;
handler = new IUEventHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
String id = iu.getPayload().get(idKey);
if (idVal.equals(id))
{
queue.offer(iu);
}
}
}, ImmutableSet.of(category));
inBuffer.registerHandler(handler);
}
public FutureIU(String category, String idKey, String idVal)
{
this(category, idKey, idVal, new InputBuffer("FutureIU", ImmutableSet.of(category)));
}
/**
* Closes the FutureIU, use only if get is not used.
*/
public void cleanup()
{
inBuffer.removeHandler(handler);
if (inBuffer.getOwningComponentName().equals("FutureIU"))
{
inBuffer.close();
}
}
/**
* Waits (if necessary) for the IU and take it (can be done only once)
*/
public AbstractIU take() throws InterruptedException
{
AbstractIU iu;
try
{
iu = queue.take();
}
finally
{
cleanup();
}
return iu;
}
/**
* Wait for at most the given time for the IU and take it (can be done only once), return null on timeout
*/
public AbstractIU take(long timeout, TimeUnit unit) throws InterruptedException
{
AbstractIU iu;
try
{
iu = queue.poll(timeout, unit);
}
finally
{
cleanup();
}
return iu;
}
}
package ipaaca.util.communication;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
/**
* Obtain multiple future ius on an specific category. Usage:<br>
* FutureIUs futures = new FutureIUs(componentx, key);<br>
* [make componentx send a IU with key=keyvaluedesired1]<br>
* AbstractIU iu = futures.take(keyvaluedesired1);<br>
* [make componentx send a IU with key=keyvaluedesired2]
* AbstractIU iu = futures.take(keyvaluedesired2);<br>
* ...<br>
* futures.close();
* @author hvanwelbergen
*/
public class FutureIUs
{
private final InputBuffer inBuffer;
private final Map<String,BlockingQueue<AbstractIU>> resultsMap = Collections.synchronizedMap(new HashMap<String,BlockingQueue<AbstractIU>>());
public FutureIUs(String category, final String idKey)
{
inBuffer = new InputBuffer("FutureIUs", ImmutableSet.of(category));
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
String id = iu.getPayload().get(idKey);
resultsMap.putIfAbsent(id, new ArrayBlockingQueue<AbstractIU>(1));
resultsMap.get(id).offer(iu);
}
}, ImmutableSet.of(category));
}
/**
* Waits (if necessary) for the IU and take it (can be done only once)
*/
public AbstractIU take(String idValue) throws InterruptedException
{
resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
return resultsMap.get(idValue).take();
}
/**
* Wait for at most the given time for the IU and take it (can be done only once), return null on timeout
*/
public AbstractIU take(String idValue, long timeout, TimeUnit unit) throws InterruptedException
{
resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
return resultsMap.get(idValue).poll(timeout, unit);
}
public void close()
{
inBuffer.close();
}
}
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaacademo;
import java.io.BufferedInputStream;
......
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaacademo;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventHandler;
import ipaaca.IUEventType;
import ipaaca.Initializer;
import ipaaca.InputBuffer;
import ipaaca.OutputBuffer;
import ipaaca.RemotePushIU;
import java.util.EnumSet;
import java.util.Set;
import javax.swing.JFrame;
import javax.swing.JLabel;
import com.google.common.collect.ImmutableSet;
public class TestListener
{
private static final class MyEventHandler implements HandlerFunctor
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
switch(type)
{
case ADDED: System.out.println("IU added "+iu.getPayload().get("CONTENT")); break;
case COMMITTED: System.out.println("IU committed"); break;
case UPDATED: System.out.println("IU updated "+iu.getPayload()); break;
case LINKSUPDATED: System.out.println("IU links updated"); break;
case RETRACTED: break;
case DELETED: break;
}
}
}
static
{
Initializer.initializeIpaacaRsb();
}
private static final String CATEGORY = "spam";
private static final double RATE = 0.5;
private UpdateThread updateThread;
public TestListener()
{
Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();
JLabel label = new JLabel("");
updateThread = new UpdateThread(new InputBuffer("TestListener", categories),label);
}
public void start()
{
updateThread.start();
}
private static class UpdateThread extends Thread
{
private InputBuffer inBuffer;
private JLabel label;
public UpdateThread(InputBuffer inBuffer, JLabel label)
{
this.inBuffer = inBuffer;
this.label = label;
EnumSet<IUEventType> types = EnumSet.of(IUEventType.ADDED,IUEventType.COMMITTED,IUEventType.UPDATED,IUEventType.LINKSUPDATED);
Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();
MyEventHandler printingEventHandler;
printingEventHandler = new MyEventHandler();
this.inBuffer.registerHandler(new IUEventHandler(printingEventHandler,types,categories));
}
@Override
public void run()
{
System.out.println("Starting!");
}
}
public static void main(String args[])
{
TestListener tl = new TestListener();
tl.start();
}
}
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaacademo;
import ipaaca.AbstractIU;
......@@ -31,6 +63,8 @@ public class TextPrinter
case COMMITTED: System.out.println("IU committed"); break;
case UPDATED: System.out.println("IU updated "+iu.getPayload()); break;
case LINKSUPDATED: System.out.println("IU links updated"); break;
case RETRACTED: break;
case DELETED: break;
}
}
......@@ -188,7 +222,8 @@ public class TextPrinter
TextPrinter tp = new TextPrinter();
OutputBuffer outBuffer = new OutputBuffer("componentX");
//OutputBuffer outBuffer = new OutputBuffer("componentX");
/*String[] inputString = {"h","e","l","l","o"," ","w","o","r","l","d","!"};
LocalIU predIU = null;
for(String str:inputString)
......