ipaacalib/java/src/ipaaca/util/BlackboardUpdateListener.java

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

ipaacalib/java/src/ipaaca/util/ComponentNotifier.java

+/*
+ * This file is part of IPAACA, the
+ *  "Incremental Processing Architecture
+ *   for Artificial Conversational Agents".  
+ *
+ * Copyright (c) 2009-2013 Sociable Agents Group
+ *                         CITEC, Bielefeld University   
+ *
+ * http://opensource.cit-ec.de/projects/ipaaca/
+ * http://purl.org/net/ipaaca
+ *
+ * This file may be licensed under the terms of of the
+ * GNU Lesser General Public License Version 3 (the ``LGPL''),
+ * or (at your option) any later version.
+ *
+ * Software distributed under the License is distributed
+ * on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+ * express or implied. See the LGPL for the specific language
+ * governing rights and limitations.
+ *
+ * You should have received a copy of the LGPL along with this
+ * program. If not, go to http://www.gnu.org/licenses/lgpl.html
+ * or write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.  
+ *
+ * The development of this software was supported by the
+ * Excellence Cluster EXC 277 Cognitive Interaction Technology.
+ * The Excellence Cluster EXC 277 is a grant of the Deutsche
+ * Forschungsgemeinschaft (DFG) in the context of the German
+ * Excellence Initiative.
+ */
+
+package ipaaca.util;
+
+import ipaaca.AbstractIU;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventHandler;
+import ipaaca.IUEventType;
+import ipaaca.InputBuffer;
+import ipaaca.LocalMessageIU;
+import ipaaca.OutputBuffer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * Utility class to handle component notification: a componentNotify is sent at initialization and whenever new components sent
+ * their componentNotify.
+ * @author hvanwelbergen
+ * 
+ */
+public class ComponentNotifier
+{
+    public static final String NOTIFY_CATEGORY = "componentNotify";
+    public static final String SEND_CATEGORIES = "send_categories";
+    public static final String RECEIVE_CATEGORIES = "recv_categories";
+    public static final String STATE = "state";
+    public static final String NAME = "name";
+    public static final String FUNCTION = "function";
+    private final OutputBuffer outBuffer;
+    private final String componentName;
+    private final String componentFunction;
+    private final ImmutableSet<String> sendCategories;
+    private final ImmutableSet<String> receiveCategories;
+    private final InputBuffer inBuffer;
+    private List<HandlerFunctor> handlers = Collections.synchronizedList(new ArrayList<HandlerFunctor>());
+    private volatile boolean isInitialized = false;
+    private final BlockingQueue<String> receiverQueue = new LinkedBlockingQueue<String>();
+    
+    private class ComponentNotifyHandler implements HandlerFunctor
+    {
+        @Override
+        public void handle(AbstractIU iu, IUEventType type, boolean local)
+        {
+            if(iu.getPayload().get(NAME).equals(componentName))return; //don't re-notify self
+            String receivers[] = iu.getPayload().get(RECEIVE_CATEGORIES).split("\\s*,\\s*");
+            receiverQueue.addAll(ImmutableSet.copyOf(receivers));
+            synchronized(handlers)
+            {
+                for (HandlerFunctor h : handlers)
+                {
+                    h.handle(iu, type, local);
+                }
+            }
+            if (iu.getPayload().get(STATE).equals("new"))
+            {
+                submitNotify(false);
+            }
+        }
+    }
+
+    /**
+     * Register a handler that will be called whenever a new component notifies this ComponentNotifier
+     */
+    public void addNotificationHandler(HandlerFunctor h)
+    {
+        handlers.add(h);
+    }
+    
+    /**
+     * Wait until the receivers are registered for categories
+     */
+    public void waitForReceivers(ImmutableSet<String> categories)
+    {
+        Set<String> unhandledCategories = new HashSet<String>(categories);        
+        while(!unhandledCategories.isEmpty())
+        {
+            try
+            {
+                unhandledCategories.remove(receiverQueue.take());
+                
+            }
+            catch (InterruptedException e)
+            {
+                Thread.interrupted();
+            }            
+        }
+    }
+    
+    /**
+     * Wait until receivers are registered for all categories sent by the component
+     */
+    public void waitForReceivers()
+    {
+        waitForReceivers(sendCategories);
+    }
+
+    private void submitNotify(boolean isNew)
+    {
+        LocalMessageIU notifyIU = new LocalMessageIU();
+        notifyIU.setCategory(NOTIFY_CATEGORY);
+        notifyIU.getPayload().put(NAME, componentName);
+        notifyIU.getPayload().put("function", componentFunction);
+        notifyIU.getPayload().put(SEND_CATEGORIES, Joiner.on(",").join(sendCategories));
+        notifyIU.getPayload().put(RECEIVE_CATEGORIES, Joiner.on(",").join(receiveCategories));
+        notifyIU.getPayload().put(STATE, isNew ? "new" : "old");
+        outBuffer.add(notifyIU);
+    }
+
+    public ComponentNotifier(String componentName, String componentFunction, Set<String> sendCategories, Set<String> receiveCategories,
+            OutputBuffer outBuffer, InputBuffer inBuffer)
+    {
+        this.componentName = componentName;
+        this.componentFunction = componentFunction;
+        this.sendCategories = ImmutableSet.copyOf(sendCategories);
+        this.receiveCategories = ImmutableSet.copyOf(receiveCategories);
+        this.outBuffer = outBuffer;
+        this.inBuffer = inBuffer;
+    }
+
+    public synchronized void initialize()
+    {
+        if(!isInitialized)
+        {
+            inBuffer.registerHandler(new IUEventHandler(new ComponentNotifyHandler(), EnumSet.of(IUEventType.ADDED, IUEventType.MESSAGE), ImmutableSet
+                    .of(NOTIFY_CATEGORY)));
+            submitNotify(true);
+            isInitialized = true;
+        }
+    }
+}

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();
+    }
+}

java/src/ipaacademo/PythonCall.java → ipaacalib/java/src/ipaacademo/PythonCall.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 java.io.BufferedInputStream;
@@ -6,6 +38,11 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 
+/**
+ * Demonstrates how to call a python script from java
+ * @author hvanwelbergen
+ *
+ */
 public class PythonCall
 {
 

ipaacalib/java/src/ipaacademo/TestListener.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;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventHandler;
+import ipaaca.IUEventType;
+import ipaaca.Initializer;
+import ipaaca.InputBuffer;
+import ipaaca.OutputBuffer;
+import ipaaca.RemotePushIU;
+
+import java.util.EnumSet;
+import java.util.Set;
+
+import javax.swing.JFrame;
+import javax.swing.JLabel;
+
+import com.google.common.collect.ImmutableSet;
+
+public class TestListener
+{
+
+    private static final class MyEventHandler implements HandlerFunctor
+    {
+        @Override
+        public void handle(AbstractIU iu, IUEventType type, boolean local)
+        {
+            switch(type)
+            {
+            case ADDED:  System.out.println("IU added "+iu.getPayload().get("CONTENT"));  break;
+            case COMMITTED: System.out.println("IU committed");  break;
+            case UPDATED: System.out.println("IU updated "+iu.getPayload()); break;
+            case LINKSUPDATED: System.out.println("IU links updated"); break;
+            case RETRACTED: break;
+            case DELETED: break;
+            }
+        }
+
+    }
+
+    static
+    {
+        Initializer.initializeIpaacaRsb();
+    }
+
+    private static final String CATEGORY = "spam";
+    private static final double RATE = 0.5;
+    private UpdateThread updateThread;
+
+    public TestListener()
+    {
+        Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();
+        JLabel label = new JLabel("");
+
+        updateThread = new UpdateThread(new InputBuffer("TestListener", categories),label);
+    }
+
+    public void start()
+    {
+        updateThread.start();
+    }
+
+    private static class UpdateThread extends Thread
+    {
+        private InputBuffer inBuffer;
+        private JLabel label;
+
+        public UpdateThread(InputBuffer inBuffer, JLabel label)
+        {
+            this.inBuffer = inBuffer;
+            this.label = label;
+
+            EnumSet<IUEventType> types = EnumSet.of(IUEventType.ADDED,IUEventType.COMMITTED,IUEventType.UPDATED,IUEventType.LINKSUPDATED);
+            Set<String> categories = new ImmutableSet.Builder<String>().add(CATEGORY).build();
+            MyEventHandler printingEventHandler;
+            printingEventHandler = new MyEventHandler();
+            this.inBuffer.registerHandler(new IUEventHandler(printingEventHandler,types,categories));
+        }
+
+        @Override
+        public void run()
+        {
+        	System.out.println("Starting!");
+        }
+
+    }
+
+    public static void main(String args[])
+    {
+        TestListener tl = new TestListener();
+        tl.start();
+    }
+}

java/src/ipaacademo/TextPrinter.java → 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;
+import ipaaca.HandlerFunctor;
+import ipaaca.IUEventHandler;
+import ipaaca.IUEventType;
+import ipaaca.Initializer;
+import ipaaca.InputBuffer;
+import ipaaca.OutputBuffer;
+import ipaaca.RemotePushIU;
+
 import java.util.EnumSet;
-import java.util.Map;
-import java.util.HashMap;
 import java.util.Set;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-
-import ipaaca.*;
 
 import javax.swing.JFrame;
 import javax.swing.JLabel;
@@ -28,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;
             }            
         }
         
@@ -185,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)

java/test/ivy.xml → 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);
+        }
+    }
+}

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
     {
@@ -361,5 +344,30 @@ public class ComponentCommunicationIntegrationTest
         assertThat(iuIn.getLinks("SAME_LEVEL"),containsInAnyOrder("iu7"));
     }
     
+    @Test
+    public void testPublishLinks() throws InterruptedException
+    {
+        LocalIU localIU2 = new LocalIU();
+        localIU2.setCategory(CATEGORY);
+        localIU2.getPayload().put("key1", "item2");
+        localIU.addLinks("SAME_LEVEL", ImmutableSet.of(localIU2.getUid()));
+        outBuffer.add(localIU);
+        outBuffer.add(localIU2);
+        Thread.sleep(200);
+        assertThat(localIU.getLinks("SAME_LEVEL"),containsInAnyOrder(localIU2.getUid()));
+    }
     
+    @Test
+    public void testPublishLinksRemote() throws InterruptedException
+    {
+        LocalIU localIU2 = new LocalIU();
+        localIU2.setCategory(CATEGORY);
+        localIU2.getPayload().put("key1", "item2");
+        localIU.addLinks("SAME_LEVEL", ImmutableSet.of(localIU2.getUid()));
+        outBuffer.add(localIU);
+        outBuffer.add(localIU2);
+        Thread.sleep(200);
+        AbstractIU iuIn = inBuffer.getIU(localIU.getUid());
+        assertThat(iuIn.getLinks("SAME_LEVEL"),containsInAnyOrder(localIU2.getUid()));
+    }
 }

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

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

 package ipaaca;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertNotNull;
 
 import java.util.Set;
 
@@ -14,10 +14,15 @@ import rsb.RSBException;
 
 import com.google.common.collect.ImmutableSet;
 
+/**
+ * Unit testcases for the input buffer
+ * @author hvanwelbergen
+ *
+ */
 public class InputBufferTest
 {
     private static final String COMPID = "Comp1";
-    private static final String CATEGORY = "category1";
+    private static final String CATEGORY = "testcat";
     private InputBuffer inBuffer;
 
 
@@ -42,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");

java/test/src/ipaaca/IuConverterTest.java → 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,23 +17,23 @@ 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
+ * @author hvanwelbergen
+ *
+ */
 public class IuConverterTest
 {
     private static final String CATEGORY = "category1";
@@ -45,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);
@@ -60,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)
@@ -72,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"));
@@ -109,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);        
     }
 }

java/test/src/ipaaca/JavaPythonTest.java → 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"));
+    }
 }

java/test/src/ipaaca/LocalIUTest.java → 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
+ */
 public class LocalIUTest
 {
     OutputBuffer mockBuffer = mock(OutputBuffer.class);
@@ -21,7 +30,7 @@ public class LocalIUTest
     @Test
     public void testCommit()
     {
-        LocalIU liu = new LocalIU("iu1");
+        LocalIU liu = new LocalIU();
         liu.getPayload().put("key1", "item1");
         liu.setBuffer(mockBuffer);
         liu.commit("commitWriter");
@@ -33,7 +42,7 @@ public class LocalIUTest
     @Test
     public void testSetPayloadOnUnpublishedIU()
     {
-        LocalIU liu = new LocalIU("iu1");
+        LocalIU liu = new LocalIU();
         liu.getPayload().put("key1", "item1");
         assertEquals("item1", liu.getPayload().get("key1"));
     }
@@ -41,7 +50,7 @@ public class LocalIUTest
     @Test
     public void testSetPayloadOnPublishedIU()
     {
-        LocalIU liu = new LocalIU("iu1");
+        LocalIU liu = new LocalIU();
         liu.setBuffer(mockBuffer);
         liu.getPayload().put("key1", "item1");
         assertEquals("item1", liu.getPayload().get("key1"));

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"));
+    }
+}