ipaacalib/java/ivy.xml

@@ -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

-[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

@@ -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("") // 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

 /*
  * 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.
@@ -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.
  */
@@ -53,7 +55,7 @@ public abstract class Buffer
     {
     	return owningComponentName + "ID" + uuid;
     }
-    
+
     public String getUniqueName()
     {
         return uniqueName;
@@ -85,6 +87,7 @@ public abstract class Buffer
      */
     public Buffer(String owningComponentName)
     {
+        Initializer.initializeIpaacaRsb();
         this.owningComponentName = owningComponentName;
         uniqueName = "undef-" + uuid;
     }
@@ -107,25 +110,30 @@ public abstract class Buffer
     {
         eventHandlers.add(handler);
     }
-   
+    
+    public void removeHandler(IUEventHandler handler)
+    {
+        eventHandlers.remove(handler);
+    }
+
     public void registerHandler(HandlerFunctor func) {
     	IUEventHandler handler;
     	handler = new IUEventHandler(func);
     	registerHandler(handler);
     }
-    
+
     public void registerHandler(HandlerFunctor func, Set<String> categories) {
     	IUEventHandler handler;
     	handler = new IUEventHandler(func, categories);
     	registerHandler(handler);
     }
-    
+
     public void registerHandler(HandlerFunctor func, EnumSet<IUEventType> eventTypes) {
     	IUEventHandler handler;
     	handler = new IUEventHandler(func, eventTypes);
     	registerHandler(handler);
     }
-   
+
     public void registerHandler(HandlerFunctor func, EnumSet<IUEventType> eventTypes, Set<String> categories) {
     	IUEventHandler handler;
     	handler = new IUEventHandler(func, eventTypes, categories);

ipaacalib/java/src/ipaaca/BufferConfiguration.java

+/*
+ * 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 java.util.HashSet;
+import java.util.Set;
+
+
+class BufferConfiguration {
+	String _owningComponentName;
+	Set<String> _category_interests;
+	String _channel;
+	boolean _resendActive;
+	// protected:
+	// 	IPAACA_MEMBER_VAR_EXPORT std::string _basename;
+	// 	IPAACA_MEMBER_VAR_EXPORT std::vector<std::string> _category_interests;
+	// 	IPAACA_MEMBER_VAR_EXPORT std::string _channel;
+
+	public BufferConfiguration(String owningComponentName) {
+		this._owningComponentName = owningComponentName;
+		this._channel = "default";
+		this._resendActive = false;
+		this._category_interests = new HashSet<String>();
+	}
+
+	public String getOwningComponentName() {
+		return this._owningComponentName;
+	}
+
+	public Set<String> getCategoryInterests() {
+		return this._category_interests;
+	}
+
+	public String getChannel() {
+		return this._channel;
+	}
+
+	public boolean getResendActive() {
+		return this._resendActive;
+	}
+	// public:
+	// 	IPAACA_HEADER_EXPORT inline BufferConfiguration(const std::string basename) { _basename = basename; }
+	// 	IPAACA_HEADER_EXPORT const std::string get_basename() const { return _basename; }
+	// 	IPAACA_HEADER_EXPORT const std::vector<std::string> get_category_interests() const { return _category_interests; }
+	// 	IPAACA_HEADER_EXPORT const std::string get_channel() const { return _channel; }
+
+}

ipaacalib/java/src/ipaaca/BufferConfigurationBuilder.java

+/*
+ * 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;
+
+public class BufferConfigurationBuilder extends BufferConfiguration {
+
+	public BufferConfigurationBuilder(String owningComponentName) {
+		super(owningComponentName);
+	}
+
+	public void setOwningComponentName(String owningComponentName) {
+		this._owningComponentName = owningComponentName;
+	}
+
+	public void addCategoryInterest(String category) {
+		this._category_interests.add(category);
+	}
+
+	public void setChannel(String channel) {
+		this._channel = channel;
+	}
+
+	public void setResendActive(boolean resendActive) {
+		this._resendActive = resendActive;
+	}
+
+	public BufferConfiguration getBufferConfiguration() {
+		return this;
+	}
+
+	// public:
+	// 	IPAACA_HEADER_EXPORT inline BufferConfigurationBuilder(const std::string basename):BufferConfiguration(basename) {}
+	// 	IPAACA_HEADER_EXPORT inline void set_basename(const std::string& basename)
+	// 	{
+	// 		_basename = basename;
+	// 	}
+	// 	IPAACA_HEADER_EXPORT inline void add_category_interest(const std::string& category)
+	// 	{
+	// 		_category_interests.push_back(category);
+	// 	}
+	// 	IPAACA_HEADER_EXPORT inline void set_channel(const std::string& channel)
+	// 	{
+	// 		_channel = channel;
+	// 	}
+
+	// 	IPAACA_HEADER_EXPORT const BufferConfiguration& get_BufferConfiguration() { return *this; }
+}

ipaacalib/java/src/ipaaca/IUConverter.java

 /*
  * 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("").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/java/src/ipaaca/IUResendFailedException.java

+/*
+ * 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;
+
+/**
+ * Error indicating that an IU is immutable because it has been committed to.
+ * @author hvanwelbergen
+ * 
+ */
+public class IUResendFailedException extends RuntimeException
+{
+    private static final long serialVersionUID = 1L;
+    private final AbstractIU iu;
+
+    public AbstractIU getIU()
+    {
+        return iu;
+    }
+
+    public IUResendFailedException(AbstractIU iu)
+    {
+        super("Resending IU " + iu.getUid() + " failed.");
+        this.iu = iu;
+    }
+}

ipaacalib/java/src/ipaaca/IURetractedException.java

+/*
+ * 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;
+
+/**
+ * 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;
+
+    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

@@ -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,8 +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;
 
@@ -42,31 +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()));
-
-        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

 /*
  * 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.
@@ -33,8 +33,11 @@
 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;
@@ -51,6 +54,10 @@ 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;
@@ -70,6 +77,9 @@ 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())
@@ -121,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/" + getUniqueShortName() + "/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):
@@ -162,6 +211,30 @@ public class InputBuffer extends Buffer
         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]
@@ -180,7 +253,7 @@ public class InputBuffer extends Buffer
         Listener listener;
         try
         {
-            listener = Factory.getInstance().createListener(new Scope("/ipaaca/category/" + category));
+            listener = Factory.getInstance().createListener(new Scope("/ipaaca/channel/" + this.channel + "/category/" + category));
         }
         catch (InitializeException e1)
         {
@@ -222,7 +295,7 @@ public class InputBuffer extends Buffer
         }
 
     }
-
+    
     // def _handle_iu_events(self, event):
     // '''Dispatch incoming IU events.
     //
@@ -273,7 +346,6 @@ public class InputBuffer extends Buffer
                 logger.warn("Spurious RemoteMessage event: already got this UID: "+rm.getUid());
                 return;
             }
-            
             //logger.info("Adding Message "+rm.getUid());
             messageStore.put(rm.getUid(), rm);
             //logger.info("Calling handlers for Message "+rm.getUid());
@@ -309,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());
@@ -327,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());
@@ -347,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());
@@ -355,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
+            }
         }
     }
 
@@ -376,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

@@ -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("") // 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("") // 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

@@ -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

+
 /*
  * 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   
+ * Copyright (c) 2009-2015 Social Cognitive Systems Group
+ *                         CITEC, Bielefeld University
  *
  * http://opensource.cit-ec.de/projects/ipaaca/
  * http://purl.org/net/ipaaca
@@ -21,7 +22,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.
@@ -34,6 +35,8 @@ 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;
@@ -52,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;
@@ -69,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.
@@ -90,6 +96,16 @@ 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);
 
@@ -101,6 +117,8 @@ 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)
@@ -112,39 +130,61 @@ public class OutputBuffer extends Buffer
             throw new RuntimeException(e);
         }
 
+        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 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):
@@ -171,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()))
         {
@@ -192,7 +232,7 @@ public class OutputBuffer extends Buffer
             {
                 iu.getPayload().remove(k, update.getWriterName());
             }
-            if (update.getNewItemsList().size() > 0) 
+            if (update.getNewItemsList().size() > 0)
             {
             	HashMap<String, String> payloadUpdate = new HashMap<String, String>();
 
@@ -216,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()))
         {
@@ -279,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()))
         {
@@ -305,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:
@@ -329,7 +398,7 @@ public class OutputBuffer extends Buffer
         Informer<Object> informer;
         try
         {
-            informer = Factory.getInstance().createInformer("/ipaaca/category/" + category);
+            informer = Factory.getInstance().createInformer("/ipaaca/channel/" + this.channel + "/category/" + category);
         }
         catch (InitializeException e1)
         {
@@ -337,7 +406,7 @@ public class OutputBuffer extends Buffer
         }
 
         informerStore.put(category, informer);
-        logger.info("Added informer on " + category);
+        logger.info("Added informer on channel " + this.channel + " and category " + category);
 
         try
         {
@@ -347,6 +416,10 @@ public class OutputBuffer extends Buffer
         {
             throw new RuntimeException(e);
         }
+        catch (RSBException e)
+        {
+            throw new RuntimeException(e);
+        }
         return informer;
     }
 
@@ -384,7 +457,7 @@ public class OutputBuffer extends Buffer
         Informer<Object> informer = getInformer(iu.getCategory());
         try
         {
-            informer.send(iu);
+            informer.publish(iu);
         }
         catch (RSBException e)
         {
@@ -418,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)
         {
@@ -463,7 +550,7 @@ public class OutputBuffer extends Buffer
         Informer<Object> informer = getInformer(iu.getCategory());
         try
         {
-            informer.send(update);
+            informer.publish(update);
         }
         catch (RSBException e)
         {
@@ -476,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

@@ -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)
@@ -55,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);
@@ -92,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()));
+            }
         }
     }
 
-    // def _remotely_enforced_setitem(self, k, v):
-    // """Sets an item when requested remotely."""
-    // return dict.__setitem__(self, k, v)
+    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;
+    }
+
     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);
@@ -138,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)
@@ -168,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
@@ -205,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
@@ -251,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

@@ -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

@@ -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,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);
@@ -92,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)
@@ -103,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("").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)
         {
@@ -130,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);
@@ -144,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("") // TODO: fix this, default in .proto?
+            PayloadItem newItem = PayloadItem.newBuilder().setKey(item.getKey()).setValue(item.getValue()).setType("STR")
                     .build();
             builder.addNewItems(newItem);
 
@@ -161,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)
         {
@@ -178,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);
@@ -187,28 +203,19 @@ 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
+    public void commit()
+    {
+        commit(null);
+    }
+
     @Override
     public void commit(String writerName)
     {
+        if (isRetracted())
+        {
+            throw new IURetractedException(this);
+        }
         if (isReadOnly())
         {
             throw new IUReadOnlyException(this);
@@ -219,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)
             {
@@ -239,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);
@@ -251,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()
     {
@@ -280,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)
     {
@@ -318,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)
         {
@@ -343,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);
@@ -354,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)
@@ -420,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)
     {
@@ -435,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)
         {
@@ -459,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);
@@ -466,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)
     {
@@ -492,6 +462,10 @@ public class RemotePushIU extends AbstractIU
         {
             throw new IUCommittedException(this);
         }
+        if (isRetracted())
+        {
+            throw new IURetractedException(this);
+        }
         if (isReadOnly())
         {
             throw new IUReadOnlyException(this);
@@ -509,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)
         {
@@ -527,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

+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

+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

+package ipaaca.util;
+
+public interface BlackboardUpdateListener
+{
+    void update();
+}