Initial service version, static support files

@@ -7,5 +7,14 @@
     <classpath type="jar" location="lib/*"/>
     <classpath type="jar" location="lib/*"/>
     <classpath type="jar" location="build/lib/*"/>
     <classpath type="jar" location="build/lib/*"/>
 
 
+    <service-resource type="model" loader="main" location="servicedef/services.xml"/>
+
+    <webapp name="comet-messenger"
+        title="Comet Messenger"
+        server="default-server"
+        location="webapp"
+        mount-point="/comet"
+        app-bar-display="false"/>
+
 </ofbiz-component>
 </ofbiz-component>
 
 

@@ -5,4 +5,19 @@
     <vendor>DriverUp</vendor>
     <vendor>DriverUp</vendor>
     <version>1.0</version>
     <version>1.0</version>
 
 
+    <service name="sendCometMessage" engine="java"
+     location="com.brainfood.ofbiz.CometMessengerServlet"
+     invoke="sendMessage" auth="false">
+        <attribute name="realm" type="String" mode="IN" optional="false"/>
+        <attribute name="channel" type="String" mode="IN" optional="false"/>
+        <attribute name="message" type="String" mode="IN" optional="false"/>
+    </service>
+
+    <service name="receiveCometMessage" engine="java"
+     location="com.brainfood.ofbiz.CometMessengerServlet"
+     invoke="receiveMessage" auth="false">
+        <attribute name="realm" type="String" mode="IN" optional="false"/>
+        <attribute name="channel" type="String" mode="IN" optional="false"/>
+        <attribute name="message" type="String" mode="IN" optional="false"/>
+    </service>
 </services>
 </services>

@@ -2,38 +2,257 @@ package com.brainfood.ofbiz;
 
 
 import org.apache.catalina.comet.CometEvent;
 import org.apache.catalina.comet.CometEvent;
 import org.apache.catalina.comet.CometProcessor;
 import org.apache.catalina.comet.CometProcessor;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.ArrayList;
+import java.util.Map;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.InputStream;
+
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import javax.servlet.http.HttpServletResponse;
+
 import javax.servlet.ServletException;
 import javax.servlet.ServletException;
-import java.io.IOException;
+import javax.servlet.ServletContext;
+
+import org.ofbiz.base.util.Debug;
+import org.ofbiz.base.util.UtilMisc;
+
+import org.ofbiz.service.DispatchContext;
+import org.ofbiz.service.LocalDispatcher;
+import org.ofbiz.service.ServiceContainer;
+import org.ofbiz.service.GenericServiceException;
+import org.ofbiz.service.ServiceUtil;;
+
+import org.ofbiz.entity.Delegator;
+import org.ofbiz.entity.DelegatorFactory;
 
 
 public class CometMessengerServlet extends HttpServlet implements CometProcessor {
 public class CometMessengerServlet extends HttpServlet implements CometProcessor {
 
 
+    public static final String module = CometMessengerServlet.class.getName();
     private static final Integer TIMEOUT = 60 * 1000;
     private static final Integer TIMEOUT = 60 * 1000;
 
 
+    protected String realm = "default";
+    public static ConcurrentHashMap <String, CometMessengerServlet>realms = new <String, CometMessengerServlet>ConcurrentHashMap();
+
+    protected ArrayList<HttpServletResponse> connections =
+        new ArrayList<HttpServletResponse>();
+    protected MessageSender messageSender = null;
+    private LocalDispatcher dispatcher = null;
+
     @Override
     @Override
     public void destroy() {
     public void destroy() {
+        connections.clear();
+        realms.remove(realm);
+
+        messageSender.stop();
+        messageSender = null;
     }
     }
 
 
     @Override
     @Override
     public void init() throws ServletException {
     public void init() throws ServletException {
+        Debug.logInfo("INIT Comet Servlet", module);
+        if (realms.get(realm) != null) Debug.logError("Realm [" + realm + "] already registered", module);
+
+        realms.put(realm, this);
+
+        messageSender = new MessageSender(realm);
+        Thread messageSenderThread =
+            new Thread(messageSender, "MessageSender[" + getServletContext().getContextPath() + "]");
+        messageSenderThread.setDaemon(true);
+        messageSenderThread.start();
+
+        dispatcher = getDispatcher(getServletContext());
+    }
+
+    public void error(CometEvent event, HttpServletRequest request, HttpServletResponse response) {
+        Debug.logInfo("ERROR " + event + " " + request + " " + response, module);
     }
     }
 
 
     public void event(final CometEvent event) throws IOException, ServletException {
     public void event(final CometEvent event) throws IOException, ServletException {
         HttpServletRequest request = event.getHttpServletRequest();
         HttpServletRequest request = event.getHttpServletRequest();
         HttpServletResponse response = event.getHttpServletResponse();
         HttpServletResponse response = event.getHttpServletResponse();
+        response.setContentType("application/json");
         if (event.getEventType() == CometEvent.EventType.BEGIN) {
         if (event.getEventType() == CometEvent.EventType.BEGIN) {
-            request.setAttribute("org.apache.tomcat.comet.timeout", TIMEOUT);
-            log("Begin for session: " + request.getSession(true).getId());
+            Debug.logInfo("Begin for session: " + request.getSession(true).getId(), module);
+            PrintWriter writer = response.getWriter();
+            writer.println("{\"channel\":\"system\", \"message\":\"CONNECT\"}");
+            writer.flush();
+            synchronized(connections) {
+                connections.add(response);
+            }
         } else if (event.getEventType() == CometEvent.EventType.ERROR) {
         } else if (event.getEventType() == CometEvent.EventType.ERROR) {
-            log("Error for session: " + request.getSession(true).getId());
+            Debug.logInfo("Error for session: " + request.getSession(true).getId(), module);
+            synchronized(connections) {
+                connections.remove(response);
+            }
             event.close();
             event.close();
         } else if (event.getEventType() == CometEvent.EventType.END) {
         } else if (event.getEventType() == CometEvent.EventType.END) {
-            log("End for session: " + request.getSession(true).getId());
+            Debug.logInfo("End for session: " + request.getSession(true).getId(), module);
+            PrintWriter writer = response.getWriter();
+            writer.println("{\"channel\":\"system\", \"message\":\"DISCONNECT\"}");
+            writer.flush();
+            synchronized(connections) {
+                connections.remove(response);
+            }
             event.close();
             event.close();
         } else if (event.getEventType() == CometEvent.EventType.READ) {
         } else if (event.getEventType() == CometEvent.EventType.READ) {
-            throw new UnsupportedOperationException("This servlet does not accept data");
+            InputStream is = request.getInputStream();
+            byte[] buf = new byte[512];
+            do {
+                int n = is.read(buf); //can throw an IOException
+                if (n > 0) {
+                    Debug.logInfo("Read " + n + " bytes: " + new String(buf, 0, n) 
+                            + " for session: " + request.getSession(true).getId(), module);
+                } else if (n < 0) {
+                    error(event, request, response);
+                    return;
+                }
+            } while (is.available() > 0);
+        }
+    }
+
+    protected void sendMessage(String channel, String message) {
+        messageSender.send(channel, message);
+    }
+
+    public static Map sendMessage(DispatchContext context, Map attr) {
+        String realm = (String) attr.get("realm");
+        String channel = (String) attr.get("channel");
+        String message = (String) attr.get("message");
+
+        CometMessengerServlet servlet = realms.get(realm);
+        if (servlet == null) {
+            return ServiceUtil.returnError("No such realm");
+        }
+
+        servlet.sendMessage(channel, message);
+
+        return ServiceUtil.returnSuccess();
+    }
+
+    public static Map receiveMessage(DispatchContext context, Map attr) {
+        // This is a NO-OP that exists for other services to subscribe SECAs
+        return ServiceUtil.returnSuccess();
+    }
+
+    protected static LocalDispatcher getDispatcher(ServletContext servletContext) {
+        LocalDispatcher dispatcher = (LocalDispatcher) servletContext.getAttribute("dispatcher");
+        if (dispatcher == null) {
+            Delegator delegator = getDelegator(servletContext);
+            dispatcher = makeWebappDispatcher(servletContext, delegator);
+            servletContext.setAttribute("dispatcher", dispatcher);
+        }
+        return dispatcher;
+    }
+
+    /** This method only sets up a dispatcher for the current webapp and passed in delegator, it does not save it to the ServletContext or anywhere else, just returns it */
+    public static LocalDispatcher makeWebappDispatcher(ServletContext servletContext, Delegator delegator) {
+        if (delegator == null) {
+            Debug.logInfo("[ContextFilter.init] ERROR: delegator not defined.", module);
+            return null;
+        }
+        // get the unique name of this dispatcher
+        String dispatcherName = servletContext.getInitParameter("localDispatcherName");
+
+        if (dispatcherName == null) {
+            Debug.logInfo("No localDispatcherName specified in the web.xml file", module);
+            dispatcherName = delegator.getDelegatorName();
+        }
+
+        LocalDispatcher dispatcher = ServiceContainer.getLocalDispatcher(dispatcherName, delegator);
+        if (dispatcher == null) {
+            Debug.logInfo("[ContextFilter.init] ERROR: dispatcher could not be initialized.", module);
+        }
+
+        return dispatcher;
+    }
+
+    public static Delegator getDelegator(ServletContext servletContext) {
+        Delegator delegator = (Delegator) servletContext.getAttribute("delegator");
+        if (delegator == null) {
+            String delegatorName = servletContext.getInitParameter("entityDelegatorName");
+
+            if (delegatorName == null || delegatorName.length() <= 0) {
+                delegatorName = "default";
+            }
+            if (Debug.verboseOn()) Debug.logVerbose("Setup Entity Engine Delegator with name " + delegatorName, module);
+            delegator = DelegatorFactory.getDelegator(delegatorName);
+            servletContext.setAttribute("delegator", delegator);
+            if (delegator == null) {
+                Debug.logInfo("[ContextFilter.init] ERROR: delegator factory returned null for delegatorName \"" + delegatorName + "\"", module);
+            }
+        }
+        return delegator;
+    }
+
+    public class MessageSender implements Runnable {
+        protected boolean running = true;
+        protected ArrayList<String> messages = new ArrayList<String>();
+        private String realm = null;
+
+        public MessageSender(String realm) {
+            this.realm = realm;
+        }
+
+        public void stop() {
+            running = false;
+        }
+
+        /**
+         * Add message for sending.
+         */
+        public void send(String channel, String message) {
+            synchronized (messages) {
+                messages.add("{\"realm\":\"" + realm + "\"," +
+                    "\"channel\":\"" + channel + "\"," +
+                    "\"message\":\"" + message + "\"}");
+                messages.notify();
+
+                try {
+                    dispatcher.runSync("receiveCometMessage", UtilMisc.toMap("realm", realm, "message", message, "channel", channel));
+                } catch (GenericServiceException ex) {
+                    Debug.logError(ex, module);
+                }
+            }
+        }
+
+        public void run() {
+            while (running) {
+                if (messages.size() == 0) {
+                    try {
+                        synchronized (messages) {
+                            messages.wait();
+                        }
+                    } catch (InterruptedException e) {
+                        // Ignore
+                    }
+                }
+
+                synchronized (connections) {
+                    String[] pendingMessages = null;
+                    synchronized (messages) {
+                        pendingMessages = messages.toArray(new String[0]);
+                        messages.clear();
+                    }
+                    // Send any pending message on all the open connections
+                    for (int i = 0; i < connections.size(); i++) {
+                        try {
+                            PrintWriter writer = connections.get(i).getWriter();
+                            for (int j = 0; j < pendingMessages.length; j++) {
+                                Debug.logInfo("SENDING: " + pendingMessages[j], module);
+                                writer.println(pendingMessages[j]);
+                            }
+                            writer.flush();
+                        } catch (IOException e) {
+                            Debug.logInfo("IOExeption sending message" + e, module);
+                        }
+                    }
+                }
+            }
         }
         }
     }
     }
 }
 }

+<?xml version="1.0"?>
+<!DOCTYPE web-app PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN" "http://java.sun.com/dtd/web-app_2_3.dtd">
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+<web-app>
+  <display-name>DriverUp API</display-name>
+  <description>DriverUp API</description>
+
+  <context-param>
+    <param-name>entityDelegatorName</param-name>
+    <param-value>default</param-value>
+    <description>The Name of the Entity Delegator to use, defined in entityengine.xml</description>
+  </context-param>
+  <context-param>
+    <param-name>localDispatcherName</param-name>
+    <param-value>cometmessenger</param-value>
+    <description>A unique name used to identify/recognize the local dispatcher for the Service Engine</description>
+  </context-param>
+
+  <servlet>
+    <servlet-name>CometMessengerServlet</servlet-name>
+    <display-name>CometMessengerServlet</display-name>
+    <description>Comet Messenger Servlet</description>
+    <servlet-class>com.brainfood.ofbiz.CometMessengerServlet</servlet-class>
+    <load-on-startup>1</load-on-startup>
+  </servlet>
+  <servlet-mapping>
+    <servlet-name>CometMessengerServlet</servlet-name>
+    <url-pattern>/messenger</url-pattern>
+  </servlet-mapping>
+</web-app>

+define([], function() {
+    var listeners = [];
+    var messenger = {
+        process: function() {
+            var url = "/comet/messenger"
+            var request =  new XMLHttpRequest();
+            var responseLength = 0;
+            request.open("GET", url, true);
+            request.setRequestHeader("Content-Type","application/x-javascript;");
+            request.onreadystatechange = function() {
+                if (request.readyState == 3 || request.readyState == 4) {
+                    if (request.status == 200){
+                        if (request.responseText) {
+                            var newText = request.responseText.substr(responseLength);
+                            if (newText.length > 0) {
+                                for (var i=0; i < listeners.length; i++) {
+                                   listeners[i](newText); 
+                                }
+                            }
+                            responseLength = request.responseText.length;
+                        }
+                    }
+                    if (request.readyState == 4) {
+                        if (request.status == 200) {
+                            messenger.process();
+                        } else {
+                            console.log('Error, retrying in 5 seconds');
+                            setTimeout(function() { messenger.process(); }, 5000);
+                        }
+                    }
+                }
+            };
+            request.send(null);
+        },
+        addListener: function(listener) {
+            listeners.push(listener);
+        },
+        removeListener: function(listener) {
+            var index = listeners.indexOf(listener);
+            if (index > -1) {
+                array.splice(index, 1);
+            }
+        }
+    }
+    messenger.process();
+
+    return messenger;
+});