ipaacalib/java/ivy.xml

@@ -6,7 +6,7 @@
    <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"  />

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);
 

ipaacalib/java/src/ipaaca/Buffer.java

@@ -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/IUConverter.java

@@ -88,12 +88,8 @@ 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("STR").addAllPayload(payloadItems).addAllLinks(links).build();
         String wireFormat = (accessMode == IU.AccessMode.MESSAGE) ? "ipaaca-messageiu" : "ipaaca-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,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

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

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("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,10 +283,14 @@ 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())
             	{
@@ -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".
  *
- * 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

@@ -36,7 +36,10 @@ 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;
@@ -49,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)
@@ -82,17 +85,23 @@ 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()));
+            }
         }
     }
 
@@ -136,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)

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,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();
-        IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision(getRevision())
+        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,11 +152,15 @@ 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())
         {
@@ -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);
@@ -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/BlackboardClient.java

 package ipaaca.util;
 
-import ipaaca.AbstractIU;
-import ipaaca.HandlerFunctor;
-import ipaaca.IUEventType;
-import ipaaca.InputBuffer;
-import ipaaca.LocalMessageIU;
-import ipaaca.OutputBuffer;
-import ipaaca.protobuf.Ipaaca.IU;
-
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -18,6 +10,13 @@ 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

ipaacalib/java/src/ipaaca/util/IpaacaLogger.java

+/*
+ * This file is part of IPAACA, the
+ *  "Incremental Processing Architecture
+ *   for Artificial Conversational Agents".
+ *
+ * Copyright (c) 2009-2016 Social Cognitive Systems Group
+ *                         CITEC, Bielefeld University
+ *
+ * http://opensource.cit-ec.de/projects/ipaaca/
+ * http://purl.org/net/ipaaca
+ *
+ * This file may be licensed under the terms of of the
+ * GNU Lesser General Public License Version 3 (the ``LGPL''),
+ * or (at your option) any later version.
+ *
+ * Software distributed under the License is distributed
+ * on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+ * express or implied. See the LGPL for the specific language
+ * governing rights and limitations.
+ *
+ * You should have received a copy of the LGPL along with this
+ * program. If not, go to http://www.gnu.org/licenses/lgpl.html
+ * or write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ *
+ * The development of this software was supported by the
+ * Excellence Cluster EXC 277 Cognitive Interaction Technology.
+ * The Excellence Cluster EXC 277 is a grant of the Deutsche
+ * Forschungsgemeinschaft (DFG) in the context of the German
+ * Excellence Initiative.
+ */
+
+package ipaaca.util;
+
+import ipaaca.LocalMessageIU;
+import ipaaca.OutputBuffer;
+
+import java.util.HashMap;
+import java.util.UUID;
+
+import org.apache.commons.lang.StringUtils;
+
+public class IpaacaLogger {
+
+	private static OutputBuffer ob;
+	private static final Object lock = new Object();
+
+	private static boolean SEND_IPAACA_LOGS = true;
+	private static String MODULE_NAME = "???";
+
+	private static void initializeOutBuffer() {
+		synchronized (lock) {
+			if (ob == null) {
+				ob = new OutputBuffer("LogSender");
+			}
+		}
+	}
+
+	public static void setModuleName(String name) {
+		synchronized (lock) {
+			MODULE_NAME = name;
+		}
+	}
+
+	public static void setLogFileName(String fileName, String logMode) {
+		initializeOutBuffer();
+		LocalMessageIU msg = new LocalMessageIU("log");
+		HashMap<String, String> pl = new HashMap<String, String>();
+		pl.put("cmd", "open_log_file");
+		pl.put("filename", fileName);
+		if (logMode != null) {
+			if (logMode.equals("append") ||
+					logMode.equals("overwrite") ||
+					logMode.equals("timestamp")) {
+				pl.put("existing", logMode);
+			} else {
+
+				return;
+			}
+		}
+		ob.add(msg);
+	}
+
+	public static void sendIpaacaLogs(boolean flag) {
+		synchronized (lock) {
+			SEND_IPAACA_LOGS = flag;
+		}
+	}
+
+	private static void logConsole(String level, String text, float now, String function, String thread) {
+		for(String line: text.split("\n")) {
+			   System.out.println("[" + level + "] " + thread + " " + function + " " + line);
+			   function = StringUtils.leftPad("", function.length(), ' ');
+			   thread = StringUtils.leftPad("", thread.length(), ' ');
+		}
+	}
+
+	private static void logIpaaca(String level, String text, float now, String function, String thread) {
+		initializeOutBuffer();
+		LocalMessageIU msg = new LocalMessageIU("log");
+		HashMap<String, String> pl = new HashMap<String, String>();
+		pl.put("module", MODULE_NAME);
+		pl.put("function", function);
+		pl.put("level", level);
+		pl.put("time", String.format("%.3f", now));
+		pl.put("thread", thread);
+		pl.put("uuid", UUID.randomUUID().toString());
+		pl.put("text", text);
+		msg.setPayload(pl);
+		ob.add(msg);
+	}
+
+	private static String getCallerName() {
+		String function = Thread.currentThread().getStackTrace()[3].getClassName();
+		function += "." + Thread.currentThread().getStackTrace()[3].getMethodName();
+		return function;
+	}
+
+	public static void logError(String msg) {
+		logError(msg, System.currentTimeMillis(), getCallerName());
+	}
+
+	public static void logError(String msg, float now) {
+		logError(msg, now, getCallerName());
+	}
+
+	private static void logError(String msg, float now, String callerName) {
+		String thread = Thread.currentThread().getName();
+		if (SEND_IPAACA_LOGS) {
+			logIpaaca("ERROR", msg, now, callerName, thread);
+		}
+		logConsole("ERROR", msg, now, callerName, thread);
+	}
+
+
+	public static void logWarn(String msg) {
+		logWarn(msg, System.currentTimeMillis(), getCallerName());
+	}
+
+	public static void logWarn(String msg, float now) {
+		logWarn(msg, now, getCallerName());
+	}
+
+	private static void logWarn(String msg, float now, String callerName) {
+		String thread = Thread.currentThread().getName();
+		if (SEND_IPAACA_LOGS) {
+			logIpaaca("WARN", msg, now, callerName, thread);
+		}
+		logConsole("WARN", msg, now, callerName, thread);
+	}
+
+
+	public static void logInfo(String msg) {
+		logInfo(msg, System.currentTimeMillis(), getCallerName());
+	}
+
+	public static void logInfo(String msg, float now) {
+		logInfo(msg, now, getCallerName());
+	}
+
+	private static void logInfo(String msg, float now, String callerName) {
+		String thread = Thread.currentThread().getName();
+		if (SEND_IPAACA_LOGS) {
+			logIpaaca("INFO", msg, now, callerName, thread);
+		}
+		logConsole("INFO", msg, now, callerName, thread);
+	}
+
+
+	public static void logDebug(String msg) {
+		logDebug(msg, System.currentTimeMillis(), getCallerName());
+	}
+
+	public static void logDebug(String msg, float now) {
+		logDebug(msg, now, getCallerName());
+	}
+
+	private static void logDebug(String msg, float now, String callerName) {
+		String thread = Thread.currentThread().getName();
+		if (SEND_IPAACA_LOGS) {
+			logIpaaca("DEBUG", msg, now, callerName, thread);
+		}
+		logConsole("DEBUG", msg, now, callerName, thread);
+	}
+
+}

ipaacalib/java/src/ipaaca/util/communication/FutureIU.java

+package ipaaca.util.communication;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableSet;
+
+import ipaaca.AbstractIU;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventHandler;
+import ipaaca.IUEventType;
+import ipaaca.InputBuffer;
+
+/**
+ * Obtain a IU in the future. Usage:<br>
+ * FutureIU fu = FutureIU("componentx", "status", "started"); //wait for componentx to send a message that is it fully started<br>
+ * [Start componentx, assumes that component x will send a message or other iu with status=started in the payload]<br>
+ * AbstractIU iu = fu.take(); //get the actual IU
+ * @author hvanwelbergen
+ */
+public class FutureIU
+{
+    private final InputBuffer inBuffer;
+    private final BlockingQueue<AbstractIU> queue = new ArrayBlockingQueue<AbstractIU>(1);
+    private final IUEventHandler handler;
+
+    public FutureIU(String category, final String idKey, final String idVal, InputBuffer inBuffer)
+    {
+        this.inBuffer = inBuffer;
+        handler = new IUEventHandler(new HandlerFunctor()
+        {
+            @Override
+            public void handle(AbstractIU iu, IUEventType type, boolean local)
+            {
+                String id = iu.getPayload().get(idKey);
+                if (idVal.equals(id))
+                {
+                    queue.offer(iu);
+                }
+            }
+        }, ImmutableSet.of(category));
+        inBuffer.registerHandler(handler);
+    }
+
+    public FutureIU(String category, String idKey, String idVal)
+    {
+        this(category, idKey, idVal, new InputBuffer("FutureIU", ImmutableSet.of(category)));
+    }
+
+    /**
+     * Closes the FutureIU, use only if get is not used.
+     */
+    public void cleanup()
+    {
+        inBuffer.removeHandler(handler);
+        if (inBuffer.getOwningComponentName().equals("FutureIU"))
+        {
+            inBuffer.close();
+        }
+    }
+
+    /**
+     * Waits (if necessary) for the IU and take it (can be done only once)
+     */
+    public AbstractIU take() throws InterruptedException
+    {
+        AbstractIU iu;
+        try
+        {
+            iu = queue.take();
+        }
+        finally
+        {
+            cleanup();
+        }
+        return iu;
+    }
+
+    /**
+     * Wait for at most the given time for the IU and take it (can be done only once), return null on timeout
+     */
+    public AbstractIU take(long timeout, TimeUnit unit) throws InterruptedException
+    {
+        AbstractIU iu;
+        try
+        {
+            iu = queue.poll(timeout, unit);
+        }
+        finally
+        {
+            cleanup();
+        }
+        return iu;
+    }
+}

ipaacalib/java/src/ipaaca/util/communication/FutureIUs.java

+package ipaaca.util.communication;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableSet;
+
+import ipaaca.AbstractIU;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventType;
+import ipaaca.InputBuffer;
+
+/**
+ * Obtain multiple future ius on an specific category. Usage:<br>
+ * FutureIUs futures = new FutureIUs(componentx, key);<br>
+ * [make componentx send a IU with key=keyvaluedesired1]<br>
+ * AbstractIU iu = futures.take(keyvaluedesired1);<br>
+ * [make componentx send a IU with key=keyvaluedesired2]
+ * AbstractIU iu = futures.take(keyvaluedesired2);<br>
+ * ...<br>
+ * futures.close();
+ * @author hvanwelbergen
+ */
+public class FutureIUs
+{
+    private final InputBuffer inBuffer;
+    private final Map<String,BlockingQueue<AbstractIU>> resultsMap = Collections.synchronizedMap(new HashMap<String,BlockingQueue<AbstractIU>>());
+    
+    public FutureIUs(String category, final String idKey)
+    {
+        inBuffer = new InputBuffer("FutureIUs", ImmutableSet.of(category));
+        inBuffer.registerHandler(new HandlerFunctor()
+        {
+            @Override
+            public void handle(AbstractIU iu, IUEventType type, boolean local)
+            {
+                String id = iu.getPayload().get(idKey);
+                resultsMap.putIfAbsent(id, new ArrayBlockingQueue<AbstractIU>(1));
+                resultsMap.get(id).offer(iu);
+            }
+        }, ImmutableSet.of(category));
+    }
+    
+    /**
+     * Waits (if necessary) for the IU and take it (can be done only once)     
+     */
+    public AbstractIU take(String idValue) throws InterruptedException
+    {
+        resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
+        return resultsMap.get(idValue).take();
+    }
+    
+    /**
+     * Wait for at most the given time for the IU and take it (can be done only once), return null on timeout        
+     */
+    public AbstractIU take(String idValue, long timeout, TimeUnit unit) throws InterruptedException
+    {
+        resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
+        return resultsMap.get(idValue).poll(timeout, unit);
+    }
+    
+    public void close()
+    {
+        inBuffer.close();
+    }
+}

ipaacalib/java/test/src/ipaaca/ComponentPushCommunicationIntegrationTest.java

@@ -183,7 +183,7 @@ public class ComponentPushCommunicationIntegrationTest
     	payloadUpdate.put("chunk12", "item2");
     	payloadUpdate.put("chunk13", "item3");
     	payloadUpdate.put("chunk14", "item4");
-    	int oldRev = iuIn.getRevision();
+    	long oldRev = iuIn.getRevision();
     	localIU.getPayload().merge(payloadUpdate);
     	Thread.sleep(200);
     	assertEquals(oldRev + 1, iuIn.getRevision());
@@ -197,7 +197,7 @@ public class ComponentPushCommunicationIntegrationTest
     	payloadUpdate2.put("chunk22", "item6");
     	payloadUpdate2.put("chunk13", "item3-changed");
     	payloadUpdate2.put("chunk14", "item4-changed");
-    	int oldRev2 = iuIn.getRevision();
+    	long oldRev2 = iuIn.getRevision();
     	iuIn.getPayload().merge(payloadUpdate2);
     	Thread.sleep(200);
     	assertEquals(oldRev2 + 1, localIU.getRevision());

ipaacalib/java/test/src/ipaaca/InputBufferTest.java

@@ -54,7 +54,7 @@ public class InputBufferTest
         iu.setOwnerName("owner");
         iu.setReadOnly(false);
         iu.setRevision(1);
-        informer.send(iu);
+        informer.publish(iu);
         
         Thread.sleep(1000);
         AbstractIU iuIn = inBuffer.getIU("uid1");