ipaacalib/java/src/ipaacademo/TextPrinter.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 ipaacademo;
 
 import ipaaca.AbstractIU;
@@ -31,6 +63,8 @@ public class TextPrinter
             case COMMITTED: System.out.println("IU committed");  break;
             case UPDATED: System.out.println("IU updated "+iu.getPayload()); break;
             case LINKSUPDATED: System.out.println("IU links updated"); break;
+            case RETRACTED: break;
+            case DELETED: break;
             }            
         }
         
@@ -188,7 +222,8 @@ public class TextPrinter
         TextPrinter tp = new TextPrinter();        
         
         
-        OutputBuffer outBuffer = new OutputBuffer("componentX");
+        //OutputBuffer outBuffer = new OutputBuffer("componentX");
+
         /*String[] inputString = {"h","e","l","l","o"," ","w","o","r","l","d","!"};
         LocalIU predIU = null;        
         for(String str:inputString)

ipaacalib/java/test/ivy.xml

@@ -4,5 +4,9 @@
       <dependency org="junit" name="junit"  rev="latest.release" />
       <dependency org="hamcrest" name="hamcrest-all"       rev="latest.release"  />
       <dependency org="mockito"   name="mockito-all"            rev="latest.release"  />
+      <dependency org="jboss"			name="javassist" rev="latest.release"  />
+      <dependency org="powermock"			name="powermock-mockito" rev="latest.release"  />
+      <dependency org="logback" name="logback-classic"     rev="latest.release"  />
+      <dependency org="logback" name="logback-core"        rev="latest.release"  />
    </dependencies>
 </ivy-module>

ipaacalib/java/test/src/ipaaca/ComponentMessageCommunicationIntegrationTest.java

+package ipaaca;
+
+import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.EnumSet;
+import java.util.Set;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * Test communication of the 'MESSAGE' type between IUs
+ * @author hvanwelbergen
+ * 
+ */
+public class ComponentMessageCommunicationIntegrationTest
+{
+    @BeforeClass
+    public static void setupStatic()
+    {
+        Initializer.initializeIpaacaRsb();
+    }
+
+    private OutputBuffer outBuffer;
+    private InputBuffer inBuffer;
+    private LocalMessageIU localIU;
+    private CountingEventHandler component1EventHandler;
+    private CountingEventHandler component2EventHandler;
+    private StoringEventHandler component1StoreHandler = new StoringEventHandler();
+    private StoringEventHandler component2StoreHandler = new StoringEventHandler();
+    private static final String CATEGORY = "category1";
+
+    @Before
+    public void setup()
+    {
+        outBuffer = new OutputBuffer("component1");
+
+        Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();
+        inBuffer = new InputBuffer("component2", categories);
+        EnumSet<IUEventType> types = EnumSet.of(IUEventType.ADDED, IUEventType.COMMITTED, IUEventType.UPDATED, IUEventType.MESSAGE);
+        component2EventHandler = new CountingEventHandler();
+        component1EventHandler = new CountingEventHandler();
+        inBuffer.registerHandler(new IUEventHandler(component2EventHandler, types, categories));
+        outBuffer.registerHandler(new IUEventHandler(component1EventHandler, types, categories));
+        inBuffer.registerHandler(new IUEventHandler(component2StoreHandler, types, categories));
+        outBuffer.registerHandler(new IUEventHandler(component1StoreHandler, types, categories));
+
+        localIU = new LocalMessageIU();
+        localIU.setCategory(CATEGORY);
+        localIU.getPayload().put("key1", "item1");
+        localIU.addLinks("INIT", ImmutableSet.of("init1", "init2"));
+    }
+
+    @After
+    public void tearDown()
+    {
+        inBuffer.close();
+        outBuffer.close();
+    }
+
+    @Test
+    public void testAddedIU() throws InterruptedException
+    {
+        outBuffer.add(localIU);
+        Thread.sleep(200);
+        AbstractIU iuIn = inBuffer.getIU(localIU.getUid());
+        assertNull(iuIn);
+        assertThat(localIU.getLinks("INIT"), containsInAnyOrder("init1", "init2"));
+        assertEquals(1, component2EventHandler.getNumberOfMessageEvents(localIU.getUid()));
+        assertEquals(0, component1EventHandler.getNumberOfMessageEvents(localIU.getUid()));
+        assertEquals(1, component2EventHandler.getNumberOfMessageEvents(localIU.getUid()));
+        assertEquals(0, component1EventHandler.getNumberOfMessageEvents(localIU.getUid()));
+        assertEquals(localIU.getUid(), component2StoreHandler.getMessageIUs().get(0).getUid());
+    }
+
+    @Test
+    public void testIUCommit() throws InterruptedException
+    {
+        outBuffer.add(localIU);
+        localIU.commit();
+        Thread.sleep(200);
+        assertEquals(0, component1EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+        assertEquals(0, component2EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+        assertFalse(component2StoreHandler.getMessageIUs().get(0).isCommitted());
+    }
+
+    @Test
+    public void testIUCommitBeforePublish() throws InterruptedException
+    {
+        localIU.commit();
+        outBuffer.add(localIU);
+        Thread.sleep(200);
+        assertEquals(0, component1EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+        assertEquals(0, component2EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+        assertTrue(component2StoreHandler.getMessageIUs().get(0).isCommitted());
+    }
+
+    @Test
+    public void testIUCommitFromInputBuffer() throws InterruptedException
+    {
+        outBuffer.add(localIU);
+        Thread.sleep(200);
+        AbstractIU iuIn = component2StoreHandler.getMessageIUs().get(0);
+
+        iuIn.commit();
+        Thread.sleep(200);
+        assertFalse(localIU.isCommitted());
+        assertEquals(0, component1EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+        assertEquals(0, component2EventHandler.getNumberOfCommitEvents(localIU.getUid()));
+    }
+
+    @Test
+    public void testIUUpdate() throws InterruptedException
+    {
+        outBuffer.add(localIU);
+        Thread.sleep(200);
+        AbstractIU iuIn = component2StoreHandler.getMessageIUs().get(0);
+        assertNull(iuIn.getPayload().get("key2"));
+
+        localIU.getPayload().put("key2", "value2");
+        Thread.sleep(200);
+        assertEquals(null, iuIn.getPayload().get("key2"));
+        assertEquals(0, component2EventHandler.getNumberOfUpdateEvents(localIU.getUid()));
+        assertEquals(0, component1EventHandler.getNumberOfUpdateEvents(localIU.getUid()));
+    }
+
+    @Test
+    public void testIUUpdateBeforePublish() throws InterruptedException
+    {
+        localIU.getPayload().put("key2", "value2");
+        outBuffer.add(localIU);
+
+        Thread.sleep(200);
+        AbstractIU iuIn = component2StoreHandler.getMessageIUs().get(0);
+        assertEquals("value2", iuIn.getPayload().get("key2"));
+    }
+
+    private void fillBuffer(int val)
+    {
+        LocalMessageIU iu = new LocalMessageIU();
+        StringBuffer buffer = new StringBuffer();
+        for (int i = 0; i < 1000000; i++)
+        {
+            buffer.append(val);
+        }
+        iu.setCategory("");
+        iu.getPayload().put("x", buffer.toString());
+        outBuffer.add(iu);
+    }
+
+    @Test
+    public void testManyMessages()
+    {
+        for (int i = 0; i < 1000; i++)
+        {
+            fillBuffer(i);
+        }
+    }
+}

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

 package ipaaca;
 
-import static org.junit.Assert.*;
+import static ipaaca.IUTestUtil.assertEqualIU;
+import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import java.util.EnumSet;
-import java.util.Map;
 import java.util.HashMap;
 import java.util.Set;
 
-import static org.hamcrest.collection.IsIterableContainingInAnyOrder.*;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -16,15 +21,13 @@ import org.junit.Test;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 
-import static ipaaca.IUTestUtil.*;
-
 /**
  * Integration test cases for IPAACA.
  * Requires a running spread daemon.
  * @author hvanwelbergen
  * 
  */
-public class ComponentCommunicationIntegrationTest
+public class ComponentPushCommunicationIntegrationTest
 {
     @BeforeClass
     public static void setupStatic()
@@ -35,66 +38,10 @@ public class ComponentCommunicationIntegrationTest
     private OutputBuffer outBuffer;
     private InputBuffer inBuffer;
     private LocalIU localIU;
-    private MyEventHandler component1EventHandler;    
-    private MyEventHandler component2EventHandler;
+    private CountingEventHandler component1EventHandler;    
+    private CountingEventHandler component2EventHandler;
     private static final String CATEGORY = "category1";
     
-    private static final class MyEventHandler implements HandlerFunctor
-    {
-        private Map<String,Integer> commitEvents = new HashMap<String,Integer>();
-        private Map<String,Integer> addEvents = new HashMap<String,Integer>();
-        private Map<String,Integer> updateEvents = new HashMap<String,Integer>();
-        
-        private void updateEventMap(String key, Map<String,Integer> map)
-        {
-            int value = 0;
-            if(map.containsKey(key))
-            {
-                value = map.get(key);
-            }
-            value++;
-            map.put(key, value);
-        }
-        
-        @Override
-        public void handle(AbstractIU iu, IUEventType type, boolean local)
-        {
-            switch(type)
-            {
-            case ADDED:  updateEventMap(iu.getUid(),addEvents);  break;
-            case COMMITTED: updateEventMap(iu.getUid(),commitEvents); break;
-            case UPDATED: updateEventMap(iu.getUid(),updateEvents); break;
-            }            
-        }
-        
-        public int getNumberOfCommitEvents(String iu)
-        {
-            if(!commitEvents.containsKey(iu))
-            {
-                return 0;
-            }
-            return commitEvents.get(iu);
-        }
-        
-        public int getNumberOfAddEvents(String iu)
-        {
-            if(!addEvents.containsKey(iu))
-            {
-                return 0;
-            }
-            return addEvents.get(iu);
-        }
-        
-        public int getNumberOfUpdateEvents(String iu)
-        {
-            if(!updateEvents.containsKey(iu))
-            {
-                return 0;
-            }
-            return updateEvents.get(iu);
-        }
-    }
-    
     @Before
     public void setup()
     {
@@ -103,8 +50,8 @@ public class ComponentCommunicationIntegrationTest
         Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();        
         inBuffer = new InputBuffer("component2", categories);
         EnumSet<IUEventType> types = EnumSet.of(IUEventType.ADDED,IUEventType.COMMITTED,IUEventType.UPDATED);
-        component2EventHandler = new MyEventHandler();
-        component1EventHandler = new MyEventHandler();
+        component2EventHandler = new CountingEventHandler();
+        component1EventHandler = new CountingEventHandler();
         inBuffer.registerHandler(new IUEventHandler(component2EventHandler,types,categories));
         outBuffer.registerHandler(new IUEventHandler(component1EventHandler,types,categories));
         
@@ -224,6 +171,42 @@ public class ComponentCommunicationIntegrationTest
         assertEquals(1,component1EventHandler.getNumberOfUpdateEvents(localIU.getUid()));
     }
     
+    @Test
+    public void testSetAllPayload() throws InterruptedException
+    {
+    	outBuffer.add(localIU);
+    	Thread.sleep(200);
+    	AbstractIU iuIn = inBuffer.getIU(localIU.getUid());
+    	
+    	HashMap<String, String> payloadUpdate = new HashMap<String, String>();
+    	payloadUpdate.put("chunk11", "item1");
+    	payloadUpdate.put("chunk12", "item2");
+    	payloadUpdate.put("chunk13", "item3");
+    	payloadUpdate.put("chunk14", "item4");
+    	long oldRev = iuIn.getRevision();
+    	localIU.getPayload().merge(payloadUpdate);
+    	Thread.sleep(200);
+    	assertEquals(oldRev + 1, iuIn.getRevision());
+    	assertTrue(iuIn.getPayload().containsKey("chunk11"));
+    	assertTrue(iuIn.getPayload().containsKey("chunk12"));
+    	assertTrue(iuIn.getPayload().containsKey("chunk13"));
+    	assertTrue(iuIn.getPayload().containsKey("chunk14"));
+    	
+    	HashMap<String, String> payloadUpdate2 = new HashMap<String, String>();
+    	payloadUpdate2.put("chunk21", "item5");
+    	payloadUpdate2.put("chunk22", "item6");
+    	payloadUpdate2.put("chunk13", "item3-changed");
+    	payloadUpdate2.put("chunk14", "item4-changed");
+    	long oldRev2 = iuIn.getRevision();
+    	iuIn.getPayload().merge(payloadUpdate2);
+    	Thread.sleep(200);
+    	assertEquals(oldRev2 + 1, localIU.getRevision());
+    	assertTrue(localIU.getPayload().containsKey("chunk21"));
+    	assertTrue(localIU.getPayload().containsKey("chunk22"));
+    	assertEquals("item3-changed", localIU.getPayload().get("chunk13"));
+    	assertEquals("item4-changed", localIU.getPayload().get("chunk14"));
+    }
+    
     @Test
     public void testIUUpdateFromInputBuffer() throws InterruptedException
     {

ipaacalib/java/test/src/ipaaca/CountingEventHandler.java

+package ipaaca;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Counts how often and in what order certain events occur
+ * @author hvanwelbergen
+ *
+ */
+final class CountingEventHandler implements HandlerFunctor
+{
+    private Map<String,Integer> messageEvents = new HashMap<String,Integer>();
+    private Map<String,Integer> commitEvents = new HashMap<String,Integer>();
+    private Map<String,Integer> addEvents = new HashMap<String,Integer>();
+    private Map<String,Integer> updateEvents = new HashMap<String,Integer>();
+    
+    private void updateEventMap(String key, Map<String,Integer> map)
+    {
+        int value = 0;
+        if(map.containsKey(key))
+        {
+            value = map.get(key);
+        }
+        value++;
+        map.put(key, value);
+    }
+    
+    @Override
+    public void handle(AbstractIU iu, IUEventType type, boolean local)
+    {
+        switch(type)
+        {
+        case ADDED:  updateEventMap(iu.getUid(),addEvents);  break;
+        case COMMITTED: updateEventMap(iu.getUid(),commitEvents); break;
+        case UPDATED: updateEventMap(iu.getUid(),updateEvents); break;
+        case MESSAGE: updateEventMap(iu.getUid(),messageEvents);break;
+        case DELETED:
+            break;
+        case LINKSUPDATED:
+            break;
+        case RETRACTED:
+            break;
+        default:
+            break;
+        }            
+    }
+    
+    public int getNumberOfCommitEvents(String iu)
+    {
+        if(!commitEvents.containsKey(iu))
+        {
+            return 0;
+        }
+        return commitEvents.get(iu);
+    }
+    
+    public int getNumberOfAddEvents(String iu)
+    {
+        if(!addEvents.containsKey(iu))
+        {
+            return 0;
+        }
+        return addEvents.get(iu);
+    }
+    
+    public int getNumberOfUpdateEvents(String iu)
+    {
+        if(!updateEvents.containsKey(iu))
+        {
+            return 0;
+        }
+        return updateEvents.get(iu);
+    }
+    
+    public int getNumberOfMessageEvents(String iu)
+    {
+        if(!messageEvents.containsKey(iu))
+        {
+            return 0;
+        }
+        return messageEvents.get(iu);
+    }
+}
\ No newline at end of file

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

 package ipaaca;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertNotNull;
 
 import java.util.Set;
 
@@ -22,7 +22,7 @@ import com.google.common.collect.ImmutableSet;
 public class InputBufferTest
 {
     private static final String COMPID = "Comp1";
-    private static final String CATEGORY = "category1";
+    private static final String CATEGORY = "testcat";
     private InputBuffer inBuffer;
 
 
@@ -47,14 +47,14 @@ public class InputBufferTest
     @Test
     public void testHandleRemotePushEvent() throws RSBException, InterruptedException
     {
-        Informer<Object> informer = Factory.getInstance().createInformer("/ipaaca/category/"+CATEGORY);
+        Informer<Object> informer = Factory.getInstance().createInformer("/ipaaca/channel/default/category/"+CATEGORY);
         informer.activate();
         RemotePushIU iu = new RemotePushIU("uid1");
-        iu.setCategory("/ipaaca/category/"+CATEGORY);
+        iu.setCategory("/ipaaca/channel/default/category/"+CATEGORY);
         iu.setOwnerName("owner");
         iu.setReadOnly(false);
         iu.setRevision(1);
-        informer.send(iu);
+        informer.publish(iu);
         
         Thread.sleep(1000);
         AbstractIU iuIn = inBuffer.getIU("uid1");

ipaacalib/java/test/src/ipaaca/IuConverterTest.java

 package ipaaca;
 
+import static ipaaca.IUTestUtil.assertEqualIU;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import ipaaca.protobuf.Ipaaca;
 import ipaaca.protobuf.Ipaaca.IU;
 import ipaaca.protobuf.Ipaaca.IU.AccessMode;
@@ -10,22 +17,17 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.*;
-import static org.mockito.Mockito.*;
-
 import org.junit.Before;
 import org.junit.Test;
 
-import com.google.common.collect.ImmutableSet;
-import com.google.protobuf.InvalidProtocolBufferException;
-
 import rsb.converter.ConversionException;
 import rsb.converter.ConverterSignature;
 import rsb.converter.UserData;
 import rsb.converter.WireContents;
 import rsb.patterns.RemoteServer;
-import static ipaaca.IUTestUtil.*;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.protobuf.InvalidProtocolBufferException;
 
 /**
  * Unit test cases for the IUConverter
@@ -50,7 +52,7 @@ public class IuConverterTest
     
 
     @Test
-    public void testSerialize() throws ConversionException, InvalidProtocolBufferException
+    public void testSerializePushIU() throws ConversionException, InvalidProtocolBufferException
     {
         RemotePushIU rpIU = new RemotePushIU("iu1");
         rpIU.setRevision(1);
@@ -65,7 +67,28 @@ public class IuConverterTest
                 
         WireContents<ByteBuffer> wiu = converter.serialize(RemotePushIU.class,rpIU);
         IU iu = IU.newBuilder().mergeFrom(wiu.getSerialization().array()).build();
-        assertEqualIU(iu, rpIU);
+        assertEqualIU(iu, rpIU);        
+        assertEquals(IU.AccessMode.PUSH,iu.getAccessMode());
+    }
+    
+    @Test
+    public void testSerializeMessageIU() throws ConversionException, InvalidProtocolBufferException
+    {
+        RemoteMessageIU rmIU = new RemoteMessageIU("iu1");
+        rmIU.setRevision(1);
+        rmIU.setOwnerName("owner");
+        rmIU.setCategory(CATEGORY);
+        rmIU.setBuffer(mockInputBuffer);
+        rmIU.getPayload().enforcedSetItem("key1", "value1");
+        rmIU.getPayload().enforcedSetItem("key2", "value2");
+        rmIU.getPayload().enforcedSetItem("key3", "value3");
+        rmIU.setLinksLocally("SAME_LEVEL",ImmutableSet.of("sibling1","sibling2"));
+        rmIU.setLinksLocally("GROUNDED_IN",ImmutableSet.of("parent1","parent2"));
+        WireContents<ByteBuffer> wiu = converter.serialize(RemoteMessageIU.class,rmIU);
+        IU iu = IU.newBuilder().mergeFrom(wiu.getSerialization().array()).build();
+        assertEqualIU(iu, rmIU);
+        assertEquals(IU.AccessMode.MESSAGE,iu.getAccessMode());
+        
     }
 
     public PayloadItem createPayloadItem(String key, String value)
@@ -77,8 +100,9 @@ public class IuConverterTest
                 .build();        
     }
     
+    
     @Test
-    public void testDeSerialize() throws ConversionException
+    public void testDeSerializePushIU() throws ConversionException
     {
         List<PayloadItem> payload = new ArrayList<PayloadItem>();
         payload.add(createPayloadItem("key1","value1"));
@@ -114,6 +138,47 @@ public class IuConverterTest
         
         assertThat(data.getData(), instanceOf(RemotePushIU.class));
         RemotePushIU rpIU = (RemotePushIU) data.getData();
-        assertEqualIU(iu, rpIU);
+        assertEqualIU(iu, rpIU);       
+        
+    }
+    
+    @Test
+    public void testDeSerializeMessageIU() throws ConversionException
+    {
+        List<PayloadItem> payload = new ArrayList<PayloadItem>();
+        payload.add(createPayloadItem("key1","value1"));
+        payload.add(createPayloadItem("key2","value2"));
+        payload.add(createPayloadItem("key3","value3"));
+        List<LinkSet> links = new ArrayList<LinkSet>();
+        links.add(
+                LinkSet.newBuilder()
+                .addAllTargets(ImmutableSet.of("sibling1","sibling2"))
+                .setType("SAME_LEVEL")
+                .build()
+                );
+        links.add(
+                LinkSet.newBuilder()
+                .addAllTargets(ImmutableSet.of("parent1","parent2"))
+                .setType("GROUNDED_IN")
+                .build()
+                );
+        Ipaaca.IU iu = Ipaaca.IU.newBuilder()
+                .setUid("uid1")
+                .setRevision(1)
+                .setCommitted(false)
+                .setOwnerName("owner")
+                .setAccessMode(AccessMode.MESSAGE)
+                .setReadOnly(false)
+                .setCategory(CATEGORY)                
+                .addAllPayload(payload)
+                .addAllLinks(links)
+                .setPayloadType("")
+                .build();
+        
+        UserData<?> data = converter.deserialize("", ByteBuffer.wrap(iu.toByteArray()));
+        
+        assertThat(data.getData(), instanceOf(RemoteMessageIU.class));
+        RemoteMessageIU rpIU = (RemoteMessageIU) data.getData();
+        assertEqualIU(iu, rpIU);        
     }
 }

ipaacalib/java/test/src/ipaaca/JavaPythonTest.java

 package ipaaca;
 
 import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import java.io.BufferedInputStream;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.util.EnumSet;
 import java.util.Set;
 
 import org.junit.Before;
@@ -24,7 +27,8 @@ import com.google.common.collect.ImmutableSet;
  */
 public class JavaPythonTest
 {
-
+    private StoringEventHandler storeHandler = new StoringEventHandler();
+    
     @BeforeClass
     public static void setupStatic()
     {
@@ -35,19 +39,20 @@ public class JavaPythonTest
 
     private static final String PYTHON_PREAMBLE = "import sys\n" 
             + "sys.path.insert(0, '../python/build/')\n"
-            + "sys.path.insert(0, '../python/lib/')\n" 
+            + "sys.path.insert(0, '../python/lib/')\n"             
+            + "sys.path.insert(0, '../../deps/python/')\n"
             + "import ipaaca, time\n";
 
     @Before
     public void setup()
     {
         Set<String> categories = new ImmutableSet.Builder<String>().add("JavaPythonTest").build();
-        inBuffer = new InputBuffer("javaside", categories);
+        inBuffer = new InputBuffer("javaside", categories);        
     }
 
-    private void printRuntimeErrors(Process p) throws IOException
+    private String getRuntimeErrors(Process p) throws IOException
     {
-
+        StringBuffer errors = new StringBuffer();
         InputStream in = p.getInputStream();
         BufferedInputStream buf = new BufferedInputStream(in);
         InputStreamReader inread = new InputStreamReader(buf);
@@ -63,12 +68,12 @@ public class JavaPythonTest
         {
             if (p.waitFor() != 0)
             {
-                System.err.println("exit value = " + p.exitValue());
+                errors.append("exit value = " + p.exitValue()+"\n");
             }
         }
         catch (InterruptedException e)
         {
-            System.err.println(e);
+            errors.append(e);
         }
 
         in = p.getErrorStream();
@@ -78,15 +83,15 @@ public class JavaPythonTest
         // Read the ls output
         while ((line = bufferedreader.readLine()) != null)
         {
-            System.out.println(line);
+            errors.append(line);
         }
+        return errors.toString();
     }
 
-    private boolean runPythonProgram(String pypr) throws IOException
+    private void runPythonProgram(String pypr) throws IOException
     {
         Process p = Runtime.getRuntime().exec(new String[] { "python", "-c", pypr });
-        printRuntimeErrors(p);
-        return p.exitValue()==0;
+        assertTrue(getRuntimeErrors(p), p.exitValue()==0);
     }
 
     @Test
@@ -99,7 +104,7 @@ public class JavaPythonTest
                 + "iu.payload = {'data':'Hello from Python!'}\n" 
                 + "time.sleep(0.1)\n" 
                 + "ob.add(iu)\n";
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
 
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
@@ -118,7 +123,7 @@ public class JavaPythonTest
                 + "time.sleep(0.1)\n"
                 + "iu.payload = {'data':'Hello from Python!'}\n";
                 
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
 
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
@@ -135,7 +140,7 @@ public class JavaPythonTest
                 +"iu.add_links('testtype',['dummy1','dummy2'])\n"
                 + "time.sleep(0.1)\n" 
                 + "ob.add(iu)\n";
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
         AbstractIU iu = inBuffer.getIUs().iterator().next();
@@ -153,7 +158,7 @@ public class JavaPythonTest
                 + "time.sleep(0.1)\n"        
                 +"iu.add_links('testtype',['dummy1','dummy2'])\n";
                 
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
         AbstractIU iu = inBuffer.getIUs().iterator().next();
@@ -169,7 +174,7 @@ public class JavaPythonTest
                 + "ob.add(iu)\n"
                 + "time.sleep(0.1)\n"
                 + "iu.commit()\n";
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
         AbstractIU iu = inBuffer.getIUs().iterator().next();
@@ -181,15 +186,31 @@ public class JavaPythonTest
     {
         String pypr = PYTHON_PREAMBLE 
                 +"ob = ipaaca.OutputBuffer('pythonside')\n"
-                +"iu = ipaaca.IU('JavaPythonTest')\n"
+                +"iu = ipaaca.IU('JavaPythonTest')\n"                
                 +"iu.commit()\n"
                 +"time.sleep(0.1)\n"
                 +"ob.add(iu)\n";
                 
-        assertTrue(runPythonProgram(pypr));
+        runPythonProgram(pypr);
         Thread.sleep(200);
         assertEquals(1, inBuffer.getIUs().size());
         AbstractIU iu = inBuffer.getIUs().iterator().next();
         assertTrue(iu.isCommitted());
     }
+    
+    @Test
+    public void testMessageFromPython()throws IOException, InterruptedException
+    {
+        inBuffer.registerHandler(new IUEventHandler(storeHandler,EnumSet.of(IUEventType.ADDED, IUEventType.MESSAGE),ImmutableSet.of("JavaPythonTest")));
+        String pypr = PYTHON_PREAMBLE 
+                +"ob = ipaaca.OutputBuffer('pythonside')\n"
+                +"iu = ipaaca.Message('JavaPythonTest')\n"
+                +"iu.payload = {'data':'Hello from Python!'}\n"
+                +"time.sleep(0.1)\n"
+                +"ob.add(iu)\n";
+        runPythonProgram(pypr);
+        Thread.sleep(200);
+        assertEquals(1,storeHandler.getMessageIUs().size());
+        assertEquals("Hello from Python!", storeHandler.getMessageIUs().get(0).getPayload().get("data"));
+    }
 }

ipaacalib/java/test/src/ipaaca/LocalIUTest.java

 package ipaaca;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
 
 import org.junit.Before;
 import org.junit.Test;
-import static org.mockito.Mockito.*;
 /**
  * Unit testcases for the LocalIU
  * @author hvanwelbergen

ipaacalib/java/test/src/ipaaca/StoringEventHandler.java

+package ipaaca;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Stores ius for which add messages occured.
+ * @author hvanwelbergen
+ * 
+ */
+public class StoringEventHandler implements HandlerFunctor
+{
+    private List<AbstractIU> addedIUs = new ArrayList<AbstractIU>();
+    private List<AbstractIU> messageIUs = new ArrayList<AbstractIU>();
+
+    public List<AbstractIU> getAddedIUs()
+    {
+        return addedIUs;
+    }
+    
+    public List<AbstractIU> getMessageIUs()
+    {
+        return messageIUs;
+    }
+
+    @Override
+    public void handle(AbstractIU iu, IUEventType type, boolean local)
+    {
+        switch (type)
+        {
+        case ADDED:
+            addedIUs.add(iu);
+            break;
+        case MESSAGE:
+            messageIUs.add(iu);
+            break;
+        case COMMITTED:
+            break;
+        case UPDATED:
+            break;
+        case DELETED:
+            break;
+        case LINKSUPDATED:
+            break;
+        case RETRACTED:
+            break;
+        default:
+            break;
+        }
+    }
+
+}

ipaacalib/java/test/src/ipaaca/util/BlackboardIntegrationTest.java

+package ipaaca.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import ipaaca.Initializer;
+
+import org.junit.After;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableMap;
+
+/**
+ * Integration tests for the blackboard
+ * @author hvanwelbergen
+ */
+public class BlackboardIntegrationTest
+{
+    static
+    {
+        Initializer.initializeIpaacaRsb();
+    }
+    
+    private Blackboard bb = new Blackboard("myblackboard","blackboardx");
+    private BlackboardClient bbc;
+    
+    @After
+    public void after()
+    {
+        bb.close();
+        if(bbc!=null)
+        {
+            bbc.close();
+        }
+    }
+    
+    @Test
+    public void testGetValueFromBlackboardBeforeConnection()
+    {
+        bb.put("key1","value1");
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        assertEquals("value1", bbc.get("key1"));        
+    }
+    
+    @Test
+    public void testGetValueFromBlackboardAfterConnection() throws InterruptedException
+    {
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bb.put("key1","value1");
+        Thread.sleep(200);
+        assertEquals("value1", bbc.get("key1"));        
+    }
+    
+    @Test
+    public void testSetValueOnBlackboard() throws InterruptedException
+    {
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bbc.put("key2","value2");
+        Thread.sleep(300);
+        assertEquals("value2", bb.get("key2"));        
+    }
+    
+    @Test
+    public void testBlackboardUpdateHandler()throws InterruptedException
+    {
+        BlackboardUpdateListener mockListener = mock(BlackboardUpdateListener.class);
+        bb.addUpdateListener(mockListener);
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bbc.put("key2","value2");
+        Thread.sleep(200);
+        bb.put("key2","value3");
+        verify(mockListener,times(1)).update();
+    }
+    
+    @Test
+    public void testBlackboardClientUpdateHandler() throws InterruptedException
+    {
+        BlackboardUpdateListener mockListener = mock(BlackboardUpdateListener.class);
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bbc.addUpdateListener(mockListener);
+        bb.put("key3","value3");
+        Thread.sleep(200);
+        bbc.put("key3","value4");
+        verify(mockListener,times(2)).update();
+    }
+    
+    @Test
+    public void testSetManyValuesOnBlackboard() throws InterruptedException
+    {
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        for(int i=0;i<100;i++)
+        {
+            bbc.put("key"+i,"value"+i);
+            bb.put("key"+i,"value"+i);
+        }
+        Thread.sleep(300);
+        assertEquals("value2", bb.get("key2"));        
+        assertEquals("value3", bb.get("key3"));
+    }
+    
+    @Test
+    public void testSetValuesOnClient() throws InterruptedException
+    {
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bbc.putAll(ImmutableMap.of("key1","value1","key2","value2"));
+        Thread.sleep(200);
+        assertEquals("value1", bb.get("key1"));        
+        assertEquals("value2", bb.get("key2"));
+    }
+    
+    @Test
+    public void testSetValuesOnBlackBoard() throws InterruptedException
+    {
+        bbc = new BlackboardClient("myblackboardclient","blackboardx");
+        bbc.waitForBlackboardConnection();
+        bb.putAll(ImmutableMap.of("key1","value1","key2","value2"));
+        Thread.sleep(200);
+        assertEquals("value1", bbc.get("key1"));        
+        assertEquals("value2", bbc.get("key2"));
+    }
+}

ipaacalib/java/test/src/ipaaca/util/ComponentNotifierIntegrationTest.java

+package ipaaca.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import ipaaca.AbstractIU;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventType;
+import ipaaca.Initializer;
+import ipaaca.InputBuffer;
+import ipaaca.LocalIU;
+import ipaaca.OutputBuffer;
+
+import java.util.Set;
+
+import lombok.Getter;
+
+import org.junit.After;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * Integration test for the ComponentNotifier, connects two of them. Requires a running spread daemon.
+ * @author hvanwelbergen
+ * 
+ */
+public class ComponentNotifierIntegrationTest
+{
+    private ComponentNotifier notifier1;
+    private ComponentNotifier notifier2;
+    private InputBuffer inBuffer;
+    private OutputBuffer outBuffer;
+    private static final String OTHER_CATEGORY="OTHER";
+    static
+    {
+        Initializer.initializeIpaacaRsb();
+    }
+
+    private class MyHandlerFunctor implements HandlerFunctor
+    {
+        @Getter
+        private volatile int numCalled = 0;
+
+        @Override
+        public void handle(AbstractIU iu, IUEventType type, boolean local)
+        {
+            numCalled++;
+        }
+    }
+
+    @After
+    public void after()
+    {
+        if (inBuffer != null)
+        {
+            inBuffer.close();
+        }
+        if (outBuffer != null)
+        {
+            outBuffer.close();
+        }
+    }
+
+    private ComponentNotifier setupCompNotifier(String id, Set<String> sendList, Set<String> recvList)
+    {
+        inBuffer = new InputBuffer(id + "in", ImmutableSet.of(ComponentNotifier.NOTIFY_CATEGORY));
+        outBuffer = new OutputBuffer(id + "out");
+        return new ComponentNotifier(id, "test", ImmutableSet.copyOf(sendList), ImmutableSet.copyOf(recvList), outBuffer, inBuffer);
+    }
+
+    private ComponentNotifier setupCompNotifierWithOtherCategoryInputBuffer(String id, Set<String> sendList, Set<String> recvList)
+    {
+        inBuffer = new InputBuffer(id + "in", ImmutableSet.of(ComponentNotifier.NOTIFY_CATEGORY, OTHER_CATEGORY));
+        outBuffer = new OutputBuffer(id + "out");
+        return new ComponentNotifier(id, "test", ImmutableSet.copyOf(sendList), ImmutableSet.copyOf(recvList), outBuffer, inBuffer);
+    }
+    
+    @Test
+    public void testSelf() throws InterruptedException
+    {
+        notifier1 = setupCompNotifier("not1", ImmutableSet.of("a1", "b1"), ImmutableSet.of("a3", "b1"));
+        MyHandlerFunctor h1 = new MyHandlerFunctor();
+        notifier1.addNotificationHandler(h1);
+
+        notifier1.initialize();
+        Thread.sleep(500);
+        assertEquals(0, h1.getNumCalled());
+    }
+
+    @Test
+    public void testTwo() throws InterruptedException
+    {
+        notifier1 = setupCompNotifier("not1", ImmutableSet.of("a1", "b1"), ImmutableSet.of("a3", "b2"));
+        notifier2 = setupCompNotifier("not2", ImmutableSet.of("a2", "b2"), ImmutableSet.of("a3", "b1"));
+        MyHandlerFunctor h1 = new MyHandlerFunctor();
+        MyHandlerFunctor h2 = new MyHandlerFunctor();
+        notifier1.addNotificationHandler(h1);
+        notifier2.addNotificationHandler(h2);
+
+        notifier1.initialize();
+        Thread.sleep(500);
+        notifier2.initialize();
+        Thread.sleep(500);
+
+        assertEquals(1, h1.getNumCalled());
+        assertEquals(1, h2.getNumCalled());
+    }
+    
+    @Test
+    public void testOtherCategoryInInputBuffer() throws InterruptedException
+    {
+        notifier1 = setupCompNotifierWithOtherCategoryInputBuffer("not1", ImmutableSet.of("a1", "b1"), ImmutableSet.of("a3", "b1"));
+        MyHandlerFunctor h1 = new MyHandlerFunctor();
+        notifier1.addNotificationHandler(h1);
+        
+        OutputBuffer out = new OutputBuffer("out");
+        LocalIU iu = new LocalIU(OTHER_CATEGORY);
+        out.add(iu);
+        Thread.sleep(500);
+        assertEquals(0, h1.getNumCalled());
+        assertNotNull(inBuffer.getIU(iu.getUid()));
+    }
+}

ipaacalib/java/test/src/ipaaca/util/ComponentNotifierTest.java

+package ipaaca.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.doAnswer;
+import ipaaca.AbstractIU;
+import ipaaca.IUEventHandler;
+import ipaaca.IUEventType;
+import ipaaca.InputBuffer;
+import ipaaca.LocalIU;
+import ipaaca.OutputBuffer;
+import ipaaca.Payload;
+
+import java.util.Set;
+
+import org.hamcrest.collection.IsIterableContainingInAnyOrder;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * Unit tests for the ComponentNotifier
+ * @author hvanwelbergen
+ */
+public class ComponentNotifierTest
+{
+    private static final ImmutableSet<String> RECV_CAT = ImmutableSet.of("testrec1", "testrc2");
+    private static final ImmutableSet<String> SEND_CAT = ImmutableSet.of("testsnd1", "testsnd2", "testsnd3");
+    private OutputBuffer mockOutBuffer = mock(OutputBuffer.class);
+    private InputBuffer mockInBuffer = mock(InputBuffer.class);
+    private IUEventHandler inputHandler;
+
+    private ComponentNotifier notifier = new ComponentNotifier("testcomp", "testfunc", SEND_CAT, RECV_CAT, mockOutBuffer, mockInBuffer);
+
+    @Before
+    public void setup()
+    {
+        doAnswer(new Answer<Void>()
+        {
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable
+            {
+                IUEventHandler handler = (IUEventHandler) (invocation.getArguments()[0]);
+                inputHandler = handler;
+                return null;
+            }
+        }).when(mockInBuffer).registerHandler(any(IUEventHandler.class));
+        notifier.initialize();
+    }
+
+    @Test
+    public void testNotifyAtInit()
+    {
+        ArgumentCaptor<LocalIU> argument = ArgumentCaptor.forClass(LocalIU.class);
+        verify(mockOutBuffer).add(argument.capture());
+        LocalIU iu = argument.getValue();
+        assertEquals(ComponentNotifier.NOTIFY_CATEGORY, iu.getCategory());
+        assertEquals("new", iu.getPayload().get(ComponentNotifier.STATE));
+        assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.RECEIVE_CATEGORIES).split(",")),
+                IsIterableContainingInAnyOrder.containsInAnyOrder(RECV_CAT.toArray(new String[0])));
+        assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.SEND_CATEGORIES).split(",")),
+                IsIterableContainingInAnyOrder.containsInAnyOrder(SEND_CAT.toArray(new String[0])));
+    }
+
+    private void sendNotify(String state, Set<String> receiveCats)
+    {
+
+        AbstractIU mockIUNotify = mock(AbstractIU.class);
+        Payload mockNotifyPayload = mock(Payload.class);
+        when(mockIUNotify.getCategory()).thenReturn(ComponentNotifier.NOTIFY_CATEGORY);
+        when(mockIUNotify.getPayload()).thenReturn(mockNotifyPayload);
+        when(mockInBuffer.getIU("iuNotify")).thenReturn(mockIUNotify);
+        when(mockNotifyPayload.get(ComponentNotifier.STATE)).thenReturn(state);
+        when(mockNotifyPayload.get(ComponentNotifier.NAME)).thenReturn("namex");
+        when(mockNotifyPayload.get(ComponentNotifier.SEND_CATEGORIES)).thenReturn("");
+        when(mockNotifyPayload.get(ComponentNotifier.RECEIVE_CATEGORIES)).thenReturn(Joiner.on(",").join(receiveCats));
+        inputHandler.call(mockInBuffer, "iuNotify", false, IUEventType.ADDED, ComponentNotifier.NOTIFY_CATEGORY);
+    }
+
+    @Test
+    public void testNotifyAtNotifyNew() throws Exception
+    {
+        sendNotify("new", ImmutableSet.of("testsnd1"));
+        ArgumentCaptor<LocalIU> argument = ArgumentCaptor.forClass(LocalIU.class);
+        verify(mockOutBuffer, times(2)).add(argument.capture());
+        LocalIU iu = argument.getAllValues().get(1);
+        assertEquals("componentNotify", iu.getCategory());
+        assertEquals("old", iu.getPayload().get("state"));
+    }
+
+    @Test
+    public void testNoNotifyAtNotifyOld() throws Exception
+    {
+        sendNotify("old", ImmutableSet.of("testsnd1"));
+        ArgumentCaptor<LocalIU> argument = ArgumentCaptor.forClass(LocalIU.class);
+        verify(mockOutBuffer, times(1)).add(argument.capture());
+    }
+
+    private class WaitForFinish extends Thread
+    {
+        public volatile boolean waitFinish = false;
+
+        public void run()
+        {
+            notifier.waitForReceivers();
+            waitFinish = true;
+        }
+
+    }
+
+    @Test
+    public void testWait() throws InterruptedException
+    {
+        WaitForFinish wff = new WaitForFinish();
+        wff.start();
+        Thread.sleep(200);
+        assertFalse(wff.waitFinish);        
+        sendNotify("new",SEND_CAT);
+        Thread.sleep(200);
+        assertTrue(wff.waitFinish);
+    }
+    
+    @Test
+    public void testWaitWrongCats() throws InterruptedException
+    {
+        WaitForFinish wff = new WaitForFinish();
+        wff.start();
+        Thread.sleep(200);
+        assertFalse(wff.waitFinish);        
+        sendNotify("new",RECV_CAT);
+        Thread.sleep(200);
+        assertFalse(wff.waitFinish);
+    }
+    
+    @Test
+    public void testWaitIncremental()throws InterruptedException
+    {
+        WaitForFinish wff = new WaitForFinish();
+        wff.start();
+        Thread.sleep(200);
+        assertFalse(wff.waitFinish);        
+        sendNotify("new",ImmutableSet.of("testsnd1", "testsnd2"));
+        Thread.sleep(200);
+        assertFalse(wff.waitFinish);
+        sendNotify("new",ImmutableSet.of("testsnd3"));
+        Thread.sleep(200);
+        assertTrue(wff.waitFinish);        
+    }
+}

ipaacalib/java/test/src/ipaaca/util/communication/FutureIUTest.java

+package ipaaca.util.communication;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import ipaaca.LocalMessageIU;
+import ipaaca.OutputBuffer;
+
+/**
+ * Unit tests for the FutureIU
+ * @author hvanwelbergen
+ *
+ */
+public class FutureIUTest
+{
+    private final OutputBuffer outBuffer = new OutputBuffer("component1");
+
+    @Test(timeout = 2000)
+    public void testSendBeforeTake() throws InterruptedException
+    {
+        FutureIU fu = new FutureIU("cat1", "status", "started");
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("status", "started");
+        outBuffer.add(message);
+        assertEquals(message.getPayload(), fu.take().getPayload());
+    }
+
+    @Test(timeout = 2000)
+    public void testSendAfterTake() throws InterruptedException
+    {
+        FutureIU fu = new FutureIU("cat1", "status", "started");
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("status", "started");
+        Runnable send = () -> {
+            try
+            {
+                Thread.sleep(1000);
+            }
+            catch (Exception e)
+            {
+                throw new RuntimeException(e);
+            }
+            outBuffer.add(message);
+        };
+        new Thread(send).start();
+        assertEquals(message.getPayload(), fu.take().getPayload());
+    }
+    
+    @Test
+    public void testInvalidKeyValue() throws InterruptedException
+    {
+        FutureIU fu = new FutureIU("cat1", "status", "started");
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("status", "cancelled");
+        assertNull(fu.take(1,TimeUnit.SECONDS));
+    }
+}

ipaacalib/java/test/src/ipaaca/util/communication/FutureIUsTest.java

+package ipaaca.util.communication;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Test;
+
+import ipaaca.LocalMessageIU;
+import ipaaca.OutputBuffer;
+
+/**
+ * Unit tests for FutureIUs
+ * @author hvanwelbergen
+ *
+ */
+public class FutureIUsTest
+{
+    private FutureIUs fus = new FutureIUs("cat1","id");
+    private final OutputBuffer outBuffer = new OutputBuffer("component1");
+    
+    @After
+    public void cleanup()
+    {
+        fus.close();
+    }
+    
+    @Test(timeout = 2000)
+    public void testSendBeforeTake() throws InterruptedException
+    {
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("id", "id1");
+        outBuffer.add(message);
+        assertEquals(message.getPayload(), fus.take("id1").getPayload());
+    }
+    
+    @Test(timeout = 2000)
+    public void testSendAfterTake() throws InterruptedException
+    {
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("id", "id1");        
+        Runnable send = () -> {
+            try
+            {
+                Thread.sleep(1000);
+            }
+            catch (Exception e)
+            {
+                throw new RuntimeException(e);
+            }
+            outBuffer.add(message);
+        };
+        new Thread(send).start();
+        assertEquals(message.getPayload(), fus.take("id1").getPayload());    
+    }
+    
+    @Test
+    public void testNonMatchingKeyValue() throws InterruptedException
+    {
+        LocalMessageIU message = new LocalMessageIU("cat1");
+        message.getPayload().put("id", "id2");
+        outBuffer.add(message);
+        assertNull(fus.take("id1", 1,TimeUnit.SECONDS));
+    }
+    
+    @Test
+    public void testMultipleKeyValues() throws InterruptedException
+    {
+        LocalMessageIU message1 = new LocalMessageIU("cat1");
+        message1.getPayload().put("id", "id1");
+        LocalMessageIU message2 = new LocalMessageIU("cat1");
+        message2.getPayload().put("id", "id2");        
+        outBuffer.add(message2);
+        outBuffer.add(message1);
+        
+        assertEquals(message1.getPayload(), fus.take("id1").getPayload());
+        assertEquals(message2.getPayload(), fus.take("id2").getPayload());
+    }
+}

ipaacalib/proto/ipaaca.proto

+// This file is part of IPAACA, the
+//  "Incremental Processing Architecture
+//   for Artificial Conversational Agents".	
+//
+// Copyright (c) 2009-2022 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.
+
+syntax = "proto2";
 
 package ipaaca.protobuf;
 
+enum TransportMessageType {
+    WireTypeRESERVED = 0;
+    WireTypeIntMessage = 1;
+    WireTypeRemoteRequestResult = 2;
+    WireTypeIU = 3;
+    WireTypeMessageIU = 4; // special case on the wire (use other converter)
+    WireTypeIUPayloadUpdate = 5;
+    WireTypeIULinkUpdate = 6;
+    WireTypeIURetraction = 7;
+    WireTypeIUCommission = 8;
+    WireTypeIUResendRequest = 9;
+    WireTypeIUPayloadUpdateRequest = 100;
+    WireTypeIUCommissionRequest = 101;
+    WireTypeIULinkUpdateRequest = 102;
+}
+
+
+
+message TransportLevelWrapper {
+    required TransportMessageType transport_message_type = 1;
+    required bytes raw_message = 2;
+}
+
 message IntMessage {
-	required sint32 value = 1;
+    required sint32 value = 1;
 }
 
 message LinkSet {
-	required string type = 1;
-	repeated string targets = 2;
+    required string type = 1;
+    repeated string targets = 2;
 }
 
 message PayloadItem {
-	required string key = 1;
-	required string value = 2;
-	required string type = 3 [default = "str"];
+    required string key = 1;
+    required string value = 2;
+    required string type = 3 [default = "str"];
 }
 
 message IU {
-	enum AccessMode {
-		PUSH = 0;
-		REMOTE = 1;
-		MESSAGE = 2;
-	}
-
-	required string uid  = 1;
-	required uint32 revision = 2;
-	required string category = 3 [default = "undef"];
-	required string payload_type = 4 [default = "MAP"];
-	required string owner_name = 5;
-	required bool committed = 6 [default = false];
-	required AccessMode access_mode = 7 [default = PUSH];
-	required bool read_only = 8 [default = false];
-	repeated PayloadItem payload = 9;
-	repeated LinkSet links = 10;
+    enum AccessMode {
+        PUSH = 0;
+        REMOTE = 1;
+        MESSAGE = 2;
+    }
+
+    required string uid  = 1;
+    required uint32 revision = 2;
+    required string category = 3 [default = "undef"];
+    required string payload_type = 4 [default = "MAP"];
+    required string owner_name = 5;
+    required bool committed = 6 [default = false];
+    required AccessMode access_mode = 7 [default = PUSH];
+    required bool read_only = 8 [default = false];
+    repeated PayloadItem payload = 9;
+    repeated LinkSet links = 10;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
 }
 
 message IUPayloadUpdate {
-	required string uid = 1;
-	required uint32 revision = 2;
-	repeated PayloadItem new_items = 3;
-	repeated string keys_to_remove = 4;
-	required bool is_delta = 5 [default = false];
-	required string writer_name = 6;
+    required string uid = 1;
+    required uint32 revision = 2;
+    repeated PayloadItem new_items = 3;
+    repeated string keys_to_remove = 4;
+    required bool is_delta = 5 [default = false];
+    required string writer_name = 6;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
 }
 
 message IURetraction {
-	required string uid = 1;
-	required uint32 revision = 2;
+    required string uid = 1;
+    required uint32 revision = 2;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
 }
 
 message IUCommission {
-	required string uid = 1;
-	required uint32 revision = 2;
-	required string writer_name = 3;
+    required string uid = 1;
+    required uint32 revision = 2;
+    required string writer_name = 3;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
 }
 
 message IULinkUpdate {
-	required string uid = 1;
-	required uint32 revision = 2;
-	repeated LinkSet new_links = 3;
-	repeated LinkSet links_to_remove = 4;
-	required bool is_delta = 5 [default = false];
-	required string writer_name = 6;
+    required string uid = 1;
+    required uint32 revision = 2;
+    repeated LinkSet new_links = 3;
+    repeated LinkSet links_to_remove = 4;
+    required bool is_delta = 5 [default = false];
+    required string writer_name = 6;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
 }
 
+message IUResendRequest {
+    required string uid = 1;
+    required string hidden_scope_name = 2;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
+}
+
+// Result for remote operations (below).
+//  Used to send a raw int, which was problematic.
+//  Usually: 0 = Failed, >0 = new revision of successfully modified resource.
+message RemoteRequestResult {
+    required uint32 result = 1;
+    optional string request_uid = 100 [default = ""];
+    //optional string request_endpoint = 101 [default = ""];
+}
+
+// Remote / request versions of buffer setters:
+//  they just go with a dedicated ID
+message IUPayloadUpdateRequest {
+    required string uid = 1;
+    required uint32 revision = 2;
+    repeated PayloadItem new_items = 3;
+    repeated string keys_to_remove = 4;
+    required bool is_delta = 5 [default = false];
+    required string writer_name = 6;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
+}
+message IUCommissionRequest {
+    required string uid = 1;
+    required uint32 revision = 2;
+    required string writer_name = 3;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
+}
+
+message IULinkUpdateRequest {
+    required string uid = 1;
+    required uint32 revision = 2;
+    repeated LinkSet new_links = 3;
+    repeated LinkSet links_to_remove = 4;
+    required bool is_delta = 5 [default = false];
+    required string writer_name = 6;
+    optional string request_uid = 100 [default = ""];
+    optional string request_endpoint = 101 [default = ""];
+}
 

ipaacalib/python/build.properties

@@ -4,8 +4,8 @@ resource.path=${shared.resources}/Shared3DModels/resource;${shared.resources}/De
 rebuild.list=
 pyzip.excludes=
 run.py=run.py
-publish.resolver=soa.core.repository
+#publish.resolver=soa.core.repository
+publish.resolver=asap.sftp.publish
 dist.dir=../../dist
 deps.dir=../../deps
 extra.python.path=/vol/soa/opt64/python-spread/current/lib/python2.7/site-packages
-

ipaacalib/python/build.xml

@@ -7,7 +7,7 @@
      <exec executable="protoc">
         <arg value="--proto_path=../proto" />
         <arg value="../proto/ipaaca.proto" />
-        <arg value="--python_out=build/" />
+		<arg value="--python_out=src/ipaaca/" />
      </exec>
    </target>
 </project>

ipaacalib/python/ivy.xml

 <ivy-module version="2.0">
    <info organisation="ipaaca" module="IpaacaPython" />
    <publications>
-   <artifact type="zip" ext="zip"/>
+   <artifact type="py.zip" ext="py.zip"/>
    </publications>
    <dependencies>
-      <dependency org="google" name="protobuf" rev="latest.release"/>
-      <dependency org="rsb" name="rsb" rev="latest.release"/>
+      <dependency org="google" name="protobuf-python" rev="latest.release"/>
    </dependencies>
 
 </ivy-module>
+

ipaacalib/python/setup.py

+#!/usr/bin/env python2
+# -*- coding: utf-8 -*-
+"""
+Created on Fri Sep  9 14:12:05 2016
+
+@author: jpoeppel
+"""
+
+from setuptools import setup
+
+import os
+import sys
+import subprocess
+
+from os import path as op 
+from distutils.spawn import find_executable
+from setuptools.command.build_py import build_py
+from setuptools.command.bdist_egg import bdist_egg
+from distutils.command.build import build
+from distutils.command.sdist import sdist
+
+class ProtoBuild(build_py):
+    """
+    This command automatically compiles all .proto files with `protoc` compiler
+    and places generated files near them -- i.e. in the same directory.
+    """
+
+    def find_protoc(self):
+        "Locates protoc executable"
+
+        if 'PROTOC' in os.environ and os.path.exists(os.environ['PROTOC']):
+            protoc = os.environ['PROTOC']
+        else:
+            protoc = find_executable('protoc')
+
+        if protoc is None:
+            sys.stderr.write('protoc not found. Is protobuf-compiler installed? \n'
+                             'Alternatively, you can point the PROTOC environment variable at a local version.')
+            sys.exit(1)
+
+        return protoc
+
+    def run(self):
+        #TODO determine path automaticall
+        packagedir = "../proto"
+        print("running build proto")
+        for protofile in filter(lambda x: x.endswith('.proto'), os.listdir(packagedir)):
+            source = op.join(packagedir, protofile)
+            output = source.replace('.proto', '_pb2.py')
+
+            if (not op.exists(output) or (op.getmtime(source) > op.getmtime(output))):
+                sys.stderr.write('Protobuf-compiling ' + source + '\n')
+                subprocess.check_call([self.find_protoc(), "-I={}".format(packagedir),'--python_out=./src/ipaaca', source])
+
+class BDist_egg(bdist_egg):
+    '''
+    Simple wrapper around the normal bdist_egg command to require
+    protobuf build before normal build.
+
+    .. codeauthor:: jwienke
+
+    '''
+
+    def run(self):
+        self.run_command('build_proto')
+        bdist_egg.run(self)
+
+
+class Build(build):
+    '''
+    Simple wrapper around the normal build command to require protobuf build
+    before normal build.
+
+    .. codeauthor:: jwienke
+    '''
+
+    def run(self):
+        self.run_command('build_proto')
+        build.run(self)
+
+
+class Sdist(sdist):
+    '''
+    Simple wrapper around the normal sdist command to require protobuf build
+    before generating the source distribution..
+
+    .. codeauthor:: jwienke
+    '''
+
+    def run(self):
+        # fetch the protocol before building the source distribution so that
+        # we have a cached version and each user can rebuild the protocol
+        # with his own protobuf version
+        self.run_command('build_proto')
+        sdist.run(self)
+
+
+version = "0.1.3" #TODO determine correct version! ideally from git, maybe do something similar to rsb/setup.py
+
+        
+setup(name="ipaaca",
+      version=version,
+      author="Hendrik Buschmeier, Ramin Yaghoubzadeh, Sören Klett",
+      author_email="hbuschme@uni-bielefeld.de,ryaghoubzadeh@uni-bielefeld.de,sklett@techfak.uni-bielefeld.de",
+      license='LGPLv3+',
+      url='https://opensource.cit-ec.de/projects/ipaaca',
+      install_requires=["paho-mqtt", "six", "protobuf"],
+      packages=["ipaaca", "ipaaca.util"],
+      package_dir={"ipaaca":"src/ipaaca"},  
+
+# TODO Do we want to add ipaaca_pb2.py to the egg or as separate package?
+#      data_files=[("./ipaaca", ["ipaaca_pb2.py"])],
+                  
+#      dependency_links=[
+#          'http://www.spread.org/files/'
+#          'SpreadModule-1.5spread4.tgz#egg=SpreadModule-1.5spread4'],
+
+      cmdclass ={
+                 "build_proto": ProtoBuild,
+                 "sdist": Sdist,
+                 "build": Build,
+                 "bdist_egg":BDist_egg
+                 }
+        
+      )