@@ -0,0 +1,1774 @@
+package backtype.storm.utils;%0A%0Aimport java.util.HashMap;%0Aimport java.util.Map;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0A%0Apublic class InprocMessaging %7B%0A    private static Map%3CInteger, LinkedBlockingQueue%3CObject%3E%3E _queues = new HashMap%3CInteger, LinkedBlockingQueue%3CObject%3E%3E();%0A    private static final Object _lock = new Object();%0A    private static int port = 1;%0A    %0A    public static int acquireNewPort() %7B%0A        int ret;%0A        synchronized(_lock) %7B%0A            ret = port;%0A            port++;%0A        %7D%0A        return ret;%0A    %7D%0A    %0A    public static void sendMessage(int port, Object msg) %7B%0A        LinkedBlockingQueue%3CObject%3E queue = getQueue(port);%0A        synchronized(_lock) %7B%0A            queue.add(msg);%0A        %7D%0A    %7D%0A    %0A    public static Object takeMessage(int port) %7B%0A        LinkedBlockingQueue%3CObject%3E queue = getQueue(port);%0A        Object ret;%0A        try %7B%0A            ret = queue.take();%0A        %7D catch (InterruptedException e) %7B%0A            throw new RuntimeException(e);%0A        %7D%0A        synchronized(_lock) %7B%0A            if(queue.size()==0) %7B%0A                _queues.remove(port);%0A            %7D%0A        %7D%0A        return ret;%0A    %7D%0A%0A    public static Object pollMessage(int port) %7B%0A        LinkedBlockingQueue%3CObject%3E queue = getQueue(port);%0A        Object ret;%0A        ret = queue.poll();%0A        synchronized(_lock) %7B%0A            if(queue.size()==0) %7B%0A                _queues.remove(port);%0A            %7D%0A        %7D%0A        return ret;%0A    %7D    %0A    %0A    private static LinkedBlockingQueue%3CObject%3E getQueue(int port) %7B%0A        synchronized(_lock) %7B%0A            if(!_queues.containsKey(port)) %7B%0A              _queues.put(port, new LinkedBlockingQueue%3CObject%3E());   %0A            %7D%0A            return _queues.get(port);%0A        %7D%0A    %7D%0A%0A%7D%0A

add inprocmessaging abstraction