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
Hide whitespace changes
Inline Side-by-side
Showing
with 697 additions and 151 deletions
ipaacalib/java/ivy.xml
View file @ 460357ca
......@@ -6,8 +6,9 @@
<dependencies>
<dependency org="slf4j" name="slf4j-api" rev="latest.release" />
<dependency org="google" name="guava" rev="latest.release" />
<dependency org="google" name="protobuf-java" rev="latest.release" />
<dependency org="google" name="protobuf-java" rev="2.6.1" />
<dependency org="rsb" name="rsb" rev="latest.release" />
<dependency org="lombok" name="lombok" rev="latest.release" />
<dependency org="apache" name="commons-lang" rev="latest.release" />
</dependencies>
</ivy-module>
ipaacalib/java/rsb.cfg deleted 100644 → 0
View file @ a67669be
[transport.spread]
host = localhost # default type is string
port = 4803 # types can be specified in angle brackets
enabled = true
ipaacalib/java/src/ipaaca/AbstractIU.java
View file @ 460357ca
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.ArrayList;
......@@ -57,10 +58,13 @@ public abstract class AbstractIU
protected Payload payload;
protected String category;
protected boolean committed = false;
protected boolean retracted = false;
private String uid;
protected int revision;
protected long revision;
private boolean readOnly = false;
public abstract IU.AccessMode getAccessMode();
protected SetMultimap<String, String> links = HashMultimap.create();
private final SetMultimap<String, String> EMPTYLINKS = HashMultimap.create();
......@@ -146,7 +150,7 @@ public abstract class AbstractIU
this.readOnly = readOnly;
}
public void setRevision(int revision)
public void setRevision(long revision)
{
this.revision = revision;
}
......@@ -186,7 +190,7 @@ public abstract class AbstractIU
return buffer;
}
public int getRevision()
public long getRevision()
{
return revision;
}
......@@ -201,6 +205,11 @@ public abstract class AbstractIU
return committed;
}
public boolean isRetracted()
{
return retracted;
}
public void setBuffer(Buffer buffer)
{
this.buffer = buffer;
......@@ -218,6 +227,8 @@ public abstract class AbstractIU
public abstract void commit();
public abstract void retract();
// XXX: might not be valid for all types of IUs
public abstract void commit(String writerName);
......@@ -231,7 +242,7 @@ public abstract class AbstractIU
List<PayloadItem> items = new ArrayList<PayloadItem>();
for (Entry<String, String> entry : newPayload.entrySet())
{
PayloadItem item = PayloadItem.newBuilder().setKey(entry.getKey()).setValue(entry.getValue()).setType("str") // TODO:default type?
PayloadItem item = PayloadItem.newBuilder().setKey(entry.getKey()).setValue(entry.getValue()).setType("STR")
.build();
items.add(item);
}
......
ipaacalib/java/src/ipaaca/Buffer.java
View file @ 460357ca
......@@ -38,6 +38,8 @@ import java.util.List;
import java.util.Set;
import java.util.UUID;
import ipaaca.Initializer;
/**
* Base class for InputBuffer and OutputBuffer.
*/
......@@ -85,6 +87,7 @@ public abstract class Buffer
*/
public Buffer(String owningComponentName)
{
Initializer.initializeIpaacaRsb();
this.owningComponentName = owningComponentName;
uniqueName = "undef-" + uuid;
}
......@@ -107,6 +110,11 @@ public abstract class Buffer
{
eventHandlers.add(handler);
}
public void removeHandler(IUEventHandler handler)
{
eventHandlers.remove(handler);
}
public void registerHandler(HandlerFunctor func) {
IUEventHandler handler;
......
ipaacalib/java/src/ipaaca/BufferConfiguration.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
package ipaaca;
import java.util.HashSet;
import java.util.Set;
......@@ -50,6 +51,7 @@ class BufferConfiguration {
this._owningComponentName = owningComponentName;
this._channel = "default";
this._resendActive = false;
this._category_interests = new HashSet<String>();
}
public String getOwningComponentName() {
......
ipaacalib/java/src/ipaaca/BufferConfigurationBuilder.java
View file @ 460357ca
......@@ -33,7 +33,7 @@
package ipaaca;
class BufferConfigurationBuilder extends BufferConfiguration {
public class BufferConfigurationBuilder extends BufferConfiguration {
public BufferConfigurationBuilder(String owningComponentName) {
super(owningComponentName);
......
ipaacalib/java/src/ipaaca/IUConverter.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
......@@ -21,7 +21,7 @@
* 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.
* 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.
......@@ -55,7 +55,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
/**
* Serializes AbstractIUs into protocolbuffer IUs and vice versa.
* @author hvanwelbergen
*
*
*/
public class IUConverter implements Converter<ByteBuffer>
{
......@@ -79,7 +79,7 @@ public class IUConverter implements Converter<ByteBuffer>
List<PayloadItem> payloadItems = new ArrayList<PayloadItem>();
for (Entry<String, String> entry : iua.getPayload().entrySet())
{
payloadItems.add(PayloadItem.newBuilder().setKey(entry.getKey()).setValue(entry.getValue()).setType("str").build());
payloadItems.add(PayloadItem.newBuilder().setKey(entry.getKey()).setValue(entry.getValue()).setType("STR").build());
}
List<LinkSet> links = new ArrayList<LinkSet>();
......@@ -88,18 +88,15 @@ public class IUConverter implements Converter<ByteBuffer>
links.add(LinkSet.newBuilder().setType(entry.getKey()).addAllTargets(entry.getValue()).build());
}
IU.AccessMode accessMode = IU.AccessMode.PUSH;
if(iua instanceof RemoteMessageIU || iua instanceof LocalMessageIU)
{
accessMode = IU.AccessMode.MESSAGE;
}
IU iu = IU.newBuilder().setUid(iua.getUid()).setRevision(iua.getRevision()).setCategory(iua.getCategory())
IU.AccessMode accessMode = iua.getAccessMode();
IU iu = IU.newBuilder().setUid(iua.getUid()).setRevision((int) iua.getRevision()).setCategory(iua.getCategory())
.setOwnerName(iua.getOwnerName()).setCommitted(iua.isCommitted()).setAccessMode(accessMode)
.setReadOnly(iua.isReadOnly()).setPayloadType("MAP").addAllPayload(payloadItems).addAllLinks(links).build();
return new WireContents<ByteBuffer>(ByteBuffer.wrap(iu.toByteArray()), "ipaaca-iu");
.setReadOnly(iua.isReadOnly()).setPayloadType("STR").addAllPayload(payloadItems).addAllLinks(links).build();
String wireFormat = (accessMode == IU.AccessMode.MESSAGE) ? "ipaaca-messageiu" : "ipaaca-iu";
return new WireContents<ByteBuffer>(ByteBuffer.wrap(iu.toByteArray()), wireFormat);
}
@Override
public UserData<?> deserialize(String wireSchema, ByteBuffer buffer) throws ConversionException
{
......@@ -107,6 +104,12 @@ public class IUConverter implements Converter<ByteBuffer>
try
{
iu = IU.newBuilder().mergeFrom(buffer.array()).build();
// If there are rsb buffer read-only issues in some build, use this code instead of the above line:
//int size = buffer.capacity();
//byte[] array = new byte[size];
//buffer.get(array, 0, size);
//iu = IU.newBuilder().mergeFrom(array).build();
}
catch (InvalidProtocolBufferException e)
{
......
ipaacalib/cpp/include/ipaaca/ipaaca-json.h → ipaacalib/java/src/ipaaca/IURetractedException.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* (formerly the Sociable Agents Group)
* CITEC, Bielefeld University
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
......@@ -22,7 +21,7 @@
* 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.
* 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.
......@@ -31,20 +30,25 @@
* Excellence Initiative.
*/
#ifndef __ipaaca_json_H__
#define __ipaaca_json_H__
#include "rapidjson/document.h"
#include "rapidjson/prettywriter.h"
#include "rapidjson/filestream.h"
#include <cstdio>
// Notes:
// - From http://stackoverflow.com/questions/10426924/json-root-element
// Actually there are two different JSON specifications. RFC 4627 requires a JSON text to be
// an object or an array. ECMA-262, 5th edition, section 15.12 does not impose this restriction.
package ipaaca;
/**
* Error indicating that an IU is immutable because it has been retracted.
*
*/
public class IURetractedException extends RuntimeException
{
private static final long serialVersionUID = 1L;
private final AbstractIU iu;
#endif // __IPAACA_JSON_H__
public AbstractIU getIU()
{
return iu;
}
public IURetractedException(AbstractIU iu)
{
super("Writing to IU " + iu.getUid() + " failed -- it has been retracted.");
this.iu = iu;
}
}
ipaacalib/java/src/ipaaca/Initializer.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,9 +32,12 @@
package ipaaca;
import java.nio.ByteBuffer;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IURetraction;
import rsb.converter.ConverterSignature;
import rsb.converter.ConverterRepository;
import rsb.converter.DefaultConverterRepository;
import rsb.converter.ProtocolBufferConverter;
......@@ -43,34 +46,74 @@ import rsb.converter.ProtocolBufferConverter;
* @author hvanwelbergen
*
*/
public final class Initializer
{
private Initializer()
{
}
public final class Initializer {
private Initializer() {}
private static volatile boolean initialized = false;
public synchronized static void initializeIpaacaRsb()
{
if(initialized)return;
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new IntConverter());
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new ProtocolBufferConverter<IUCommission>(IUCommission.getDefaultInstance()));
// dlw
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new ProtocolBufferConverter<IUResendRequest>(IUResendRequest.getDefaultInstance()));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-iu", RemotePushIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-localiu", LocalIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-messageiu", RemoteMessageIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-localmessageiu", LocalMessageIU.class)));
public synchronized static void initializeIpaacaRsb() {
if (initialized)
return;
ConverterRepository<ByteBuffer> dcr =
DefaultConverterRepository.getDefaultConverterRepository();
// for IU revision numbers
dcr.addConverter(
new IntConverter());
// IU commit messages
dcr.addConverter(
new ProtocolBufferConverter<IUCommission>(
IUCommission.getDefaultInstance()));
// IU commit messages
dcr.addConverter(
new ProtocolBufferConverter<IURetraction>(
IURetraction.getDefaultInstance()));
// IU resend request messages
dcr.addConverter(
new ProtocolBufferConverter<IUResendRequest>(
IUResendRequest.getDefaultInstance()));
// IUs
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-iu",
RemotePushIU.class)));
// Local IUs
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-localiu",
LocalIU.class)));
// Messages
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-messageiu",
RemoteMessageIU.class)));
// LocalMessages
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-localmessageiu",
LocalMessageIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new PayloadConverter());
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new LinkUpdateConverter());
// Payloads
dcr.addConverter(
new PayloadConverter());
// LinkUpdates
dcr.addConverter(
new LinkUpdateConverter());
initialized = true;
}
}
ipaacalib/java/src/ipaaca/InputBuffer.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
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;
......@@ -441,6 +442,22 @@ 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());
}
}
}
}
......@@ -466,10 +483,10 @@ public class InputBuffer extends Buffer
rServer = getRemoteServer(writerName);
if ((rServer != null)&&(uid != null)) {
IUResendRequest iurr = IUResendRequest.newBuilder().setUid(uid).setHiddenScopeName(hiddenScopeName).build();
int rRevision = 0;
long rRevision = 0;
try
{
rRevision = (Integer) rServer.call("resendRequest", iurr);
rRevision = (Long) rServer.call("resendRequest", iurr);
}
catch (RSBException e)
{
......@@ -483,6 +500,10 @@ public class InputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (rRevision == 0)
{
//throw new IUResendFailedException(aiu); // TODO
......@@ -508,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();
......
ipaacalib/java/src/ipaaca/LocalIU.java
View file @ 460357ca
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate.Builder;
......@@ -50,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;
......@@ -105,6 +111,10 @@ public class LocalIU extends AbstractIU
synchronized (revisionLock)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (committed)
{
throw new IUCommittedException(this);
......@@ -121,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++;
......@@ -148,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);
......@@ -181,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());
}
}
......@@ -226,13 +256,17 @@ 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("str") // 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);
}
......@@ -249,14 +283,18 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
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") // TODO: fix this, default in .proto?
PayloadItem newItem = PayloadItem.newBuilder().setKey(item.getKey()).setValue(item.getValue()).setType("STR")
.build();
builder.addNewItems(newItem);
......@@ -270,15 +308,29 @@ public class LocalIU extends AbstractIU
@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)
{
......@@ -288,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);
}
......@@ -305,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);
}
......
ipaacalib/java/src/ipaaca/LocalMessageIU.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,6 +32,7 @@
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.
......@@ -50,4 +51,9 @@ public class LocalMessageIU extends LocalIU
super(category);
}
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.MESSAGE;
}
}
ipaacalib/java/src/ipaaca/OutputBuffer.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -34,6 +35,7 @@ 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;
......@@ -53,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;
......@@ -129,48 +133,58 @@ public class OutputBuffer extends Buffer
this.channel = ipaaca_channel;
}
private final class RemoteUpdatePayload extends 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 extends 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 extends 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 DataCallback<Integer, IUResendRequest>
private final class RemoteResendRequest extends EventCallback //DataCallback<Long, IUResendRequest>
{
@Override
public Integer invoke(IUResendRequest data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteResendRequest");
return remoteResendRequest(data);
return new Event(Long.class, new Long(remoteResendRequest((IUResendRequest) request.getData())));
}
}
// def _remote_update_payload(self, update):
......@@ -197,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()))
{
......@@ -242,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()))
{
......@@ -305,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()))
{
......@@ -334,7 +348,7 @@ public class OutputBuffer extends Buffer
/*
* Resend an requested iu over the specific hidden channel. (dlw) TODO
*/
private int remoteResendRequest(IUResendRequest iu_resend_request_pack)
private long remoteResendRequest(IUResendRequest iu_resend_request_pack)
{
if (!iuStore.containsKey(iu_resend_request_pack.getUid()))
{
......@@ -347,7 +361,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu_resend_request_pack.getHiddenScopeName());
try
{
informer.send(iu);
informer.publish(iu);
}
catch (RSBException e)
{
......@@ -402,6 +416,10 @@ public class OutputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
return informer;
}
......@@ -439,7 +457,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(iu);
informer.publish(iu);
}
catch (RSBException e)
{
......@@ -473,12 +491,26 @@ public class OutputBuffer extends Buffer
*/
protected void sendIUCommission(AbstractIU iu, String writerName)
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(iu.getUid()).setRevision(iu.getRevision())
.setWriterName(iu.getOwnerName() != null ? iu.getOwnerName() : writerName).build();
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)
{
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)
{
......@@ -518,7 +550,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......@@ -531,7 +563,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......
ipaacalib/java/src/ipaaca/Payload.java
View file @ 460357ca
......@@ -34,7 +34,12 @@ 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;
......@@ -47,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)
......@@ -80,32 +85,38 @@ 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(), pseudoConvertFromJSON(item.getValue(), item.getType()));
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")) {
System.out.println("Received JSON IU");
if (value.startsWith("\"")) {
return value.replaceAll("\\\"", "");
} 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;
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;
}
void enforcedSetItem(String key, String value)
......@@ -134,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)
......@@ -196,7 +210,7 @@ public class Payload implements Map<String, String>
public void putAll(Map<? extends String, ? extends String> newItems)
{
putAll(newItems);
putAll(newItems, null);
}
public void putAll(Map<? extends String, ? extends String> newItems, String writer)
......
ipaacalib/java/src/ipaaca/RemoteMessageIU.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.List;
......@@ -45,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);
......@@ -58,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)
{
......
ipaacalib/java/src/ipaaca/RemotePushIU.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
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;
......@@ -66,6 +67,11 @@ 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;
......@@ -84,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)
......@@ -95,20 +101,26 @@ 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("str").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)
{
......@@ -122,6 +134,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -136,15 +152,19 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
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") // TODO: fix this, default in .proto?
PayloadItem newItem = PayloadItem.newBuilder().setKey(item.getKey()).setValue(item.getValue()).setType("STR")
.build();
builder.addNewItems(newItem);
......@@ -153,10 +173,10 @@ public class RemotePushIU extends AbstractIU
RemoteServer server = getInputBuffer().getRemoteServer(this);
logger.debug("Remote server has methods {}", server.getMethods());
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
newRevision = (Long) server.call("updatePayload", update);
}
catch (RSBException e)
{
......@@ -170,6 +190,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -179,9 +203,19 @@ public class RemotePushIU extends AbstractIU
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);
......@@ -192,13 +226,13 @@ 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)
{
......@@ -212,6 +246,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUCommittedException(this);
......@@ -254,18 +292,22 @@ 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)
{
......@@ -279,6 +321,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -301,9 +347,9 @@ public class RemotePushIU extends AbstractIU
payload.enforcedRemoveItem(key);
}
for (PayloadItem item : update.getNewItemsList()) {
if (item.getType().equals("str")) {
if (item.getType().equals("STR")) {
payload.enforcedSetItem(item.getKey(), item.getValue());
} else if (item.getType().equals("json")) {
} else if (item.getType().equals("JSON")) {
String value = item.getValue();
if (value.startsWith("\"")) {
payload.enforcedSetItem(item.getKey(), value.replaceAll("\\\"", ""));
......@@ -357,6 +403,11 @@ public class RemotePushIU extends AbstractIU
committed = true;
}
public void applyRetraction()
{
retracted = true;
}
@Override
void removeFromPayload(Object key, String writer)
{
......@@ -364,17 +415,21 @@ 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)
{
......@@ -388,6 +443,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -403,6 +462,10 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
......@@ -420,11 +483,11 @@ 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)
{
......@@ -438,6 +501,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......
ipaacalib/java/src/ipaaca/util/Blackboard.java 0 → 100644
View file @ 460357ca
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();
}
}
ipaacalib/java/src/ipaaca/util/BlackboardClient.java 0 → 100644
View file @ 460357ca
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);
}
}
ipaacalib/java/src/ipaaca/util/BlackboardUpdateListener.java 0 → 100644
View file @ 460357ca
package ipaaca.util;
public interface BlackboardUpdateListener
{
void update();
}