jmsserviceengine.java

国外的一套开源CRM

            throw new GenericServiceException("JMS Internal Error.", je);
        }
        return ServiceUtil.returnSuccess();

    }

    protected Map runQueue(ModelService modelService, Map context, Element server) throws GenericServiceException {
        String serverName = server.getAttribute("jndi-server-name");
        String jndiName = server.getAttribute("jndi-name");
        String queueName = server.getAttribute("topic-queue");
        String userName = server.getAttribute("username");
        String password = server.getAttribute("password");
        String clientId = server.getAttribute("client-id");

        InitialContext jndi = null;
        QueueConnectionFactory factory = null;
        QueueConnection con = null;
        
        try {
            jndi = JNDIContextFactory.getInitialContext(serverName);
            factory = (QueueConnectionFactory) jndi.lookup(jndiName);  
        } catch (GeneralException ge){
            throw new GenericServiceException("Problems getting JNDI InitialContext.", ge.getNested());        
        } catch (NamingException ne) {
            JNDIContextFactory.clearInitialContext(serverName);           
            try {
                jndi = JNDIContextFactory.getInitialContext(serverName);
                factory = (QueueConnectionFactory) jndi.lookup(jndiName); 
            } catch (GeneralException ge2) {
                throw new GenericServiceException("Problems getting JNDI InitialContext.", ge2.getNested());            
            } catch (NamingException ne2) {
                throw new GenericServiceException("JNDI lookup problem.", ne2);
            }            
        }
        
        try {
            con = factory.createQueueConnection(userName, password);

            if (clientId != null && clientId.length() > 1)
                con.setClientID(clientId);
            con.start();

            QueueSession session = con.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = (Queue) jndi.lookup(queueName);
            QueueSender sender = session.createSender(queue);

            // create/send the message
            Message message = makeMessage(session, modelService, context);

            sender.send(message);
            if (Debug.verboseOn()) Debug.logVerbose("Sent JMS Message to " + queueName, module);

            // close the connections
            sender.close();
            session.close();
            con.close();
        } catch (NamingException ne) {
            throw new GenericServiceException("Problems with JNDI lookup.", ne);
        } catch (JMSException je) {
            throw new GenericServiceException("JMS Internal Error.", je);
        }
        return ServiceUtil.returnSuccess();
    }

    protected Map runXaQueue(ModelService modelService, Map context, Element server) throws GenericServiceException {
        String serverName = server.getAttribute("jndi-server-name");
        String jndiName = server.getAttribute("jndi-name");
        String queueName = server.getAttribute("topic-queue");
        String userName = server.getAttribute("username");
        String password = server.getAttribute("password");
        String clientId = server.getAttribute("client-id");
        
        InitialContext jndi = null;
        XAQueueConnectionFactory factory = null;
        XAQueueConnection con = null;        
        
        try {
            jndi = JNDIContextFactory.getInitialContext(serverName);
            factory = (XAQueueConnectionFactory) jndi.lookup(jndiName);                        
        } catch (GeneralException ge){
            throw new GenericServiceException("Problems getting JNDI InitialContext.", ge.getNested());
        } catch (NamingException ne) {
            JNDIContextFactory.clearInitialContext(serverName); 
            try {
                jndi = JNDIContextFactory.getInitialContext(serverName);
                factory = (XAQueueConnectionFactory) jndi.lookup(jndiName);
            } catch (GeneralException ge2){
                throw new GenericServiceException("Problems getting JNDI InitialContext.", ge2.getNested());                
            } catch (NamingException ne2) {
                throw new GenericServiceException("JNDI lookup problems.", ne2);
            }
        }
                        
        try {
            con = factory.createXAQueueConnection(userName, password);

            if (clientId != null && clientId.length() > 1)
                con.setClientID(userName);
            con.start();

            // enlist the XAResource
            XAQueueSession session = con.createXAQueueSession();
            XAResource resource = session.getXAResource();

            if (TransactionUtil.getStatus() == TransactionUtil.STATUS_ACTIVE)
                TransactionUtil.enlistResource(resource);

            Queue queue = (Queue) jndi.lookup(queueName);
            QueueSession qSession = session.getQueueSession();
            QueueSender sender = qSession.createSender(queue);

            // create/send the message
            Message message = makeMessage(session, modelService, context);

            sender.send(message);

            if (TransactionUtil.getStatus() != TransactionUtil.STATUS_ACTIVE)
                session.commit();

            Debug.logInfo("Message sent.", module);

            // close the connections
            sender.close();
            session.close();
            con.close();
        } catch (GenericTransactionException gte) {
            throw new GenericServiceException("Problems enlisting resource w/ transaction manager.", gte.getNested());
        } catch (NamingException ne) {            
            throw new GenericServiceException("Problems with JNDI lookup.", ne);
        } catch (JMSException je) {           
            throw new GenericServiceException("JMS Internal Error.", je);
        }
        return ServiceUtil.returnSuccess();
    }

    protected Map run(ModelService modelService, Map context) throws GenericServiceException {
        Element serviceElement = getServiceElement(modelService);
        List serverList = serverList(serviceElement);

        Map result = new HashMap();
        Iterator i = serverList.iterator();

        while (i.hasNext()) {
            Element server = (Element) i.next();
            String serverType = server.getAttribute("type");

            if (serverType.equals("topic"))
                result.putAll(runTopic(modelService, context, server));
            else if (serverType.equals("queue"))
                result.putAll(runQueue(modelService, context, server));
            else
                throw new GenericServiceException("Illegal server messaging type.");
        }
        return result;
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runSync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map)
     */
    public Map runSync(String localName, ModelService modelService, Map context) throws GenericServiceException {
        return run(modelService, context);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runSyncIgnore(java.lang.String, org.ofbiz.service.ModelService, java.util.Map)
     */
    public void runSyncIgnore(String localName, ModelService modelService, Map context) throws GenericServiceException {
        run(modelService, context);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runAsync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map, org.ofbiz.service.GenericRequester, boolean)
     */
    public void runAsync(String localName, ModelService modelService, Map context, GenericRequester requester, boolean persist) throws GenericServiceException {        
        Map result = run(modelService, context);

        requester.receiveResult(result);
    }

    /**
     * @see org.ofbiz.service.engine.GenericEngine#runAsync(java.lang.String, org.ofbiz.service.ModelService, java.util.Map, boolean)
     */
    public void runAsync(String localName, ModelService modelService, Map context, boolean persist) throws GenericServiceException {
        run(modelService, context);
    }

}