__init__.py

OpenWFE是一个开放源码的Java工作流引擎。它是一个完整的业务处理管理套件：一个引擎

#
# Copyright (c) 2005, John Mettraux, OpenWFE.org
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without 
# modification, are permitted provided that the following conditions are met:
# 
# . Redistributions of source code must retain the above copyright notice, this
#   list of conditions and the following disclaimer.  
# 
# . Redistributions in binary form must reproduce the above copyright notice, 
#   this list of conditions and the following disclaimer in the documentation 
#   and/or other materials provided with the distribution.
# 
# . Neither the name of the "OpenWFE" nor the names of its contributors may be
#   used to endorse or promote products derived from this software without
#   specific prior written permission.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
# POSSIBILITY OF SUCH DAMAGE.
#
# $Id: __init__.py 2142 2005-09-30 19:25:50Z jmettraux $
#

"""

    SocketDispatch and SocketListener python equivalents

    $Id: __init__.py 2142 2005-09-30 19:25:50Z jmettraux $

"""

import sys, time, re
from time import sleep
from select import select
from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from xml.dom.minidom import Document, parse, parseString
from openwfe.applic import Service, ServiceException
from openwfe.workitem import codec
from openwfe.xmlutils import getEncoding


OK_REPLY = 'ok-reply'
FATAL_REPLY = 'fatal-reply'
WARNING_REPLY = 'warning-reply'


#MAX_RETRY = 7

RE_END_OF_ITEM = re.compile('(</launchitem>|</workitem>|</cancelitem>)')


#
# REPLIES

class ListenerReply: pass


class OkReply (ListenerReply): pass


class WarningReply (ListenerReply):

    message = None
    exceptionMessage = None
    exceptionStackTrace = None

    def __init__ (self, message):
        self.message = message
    
    def __init__ (self, message, exceptionMessage, exceptionStackTrace):
        self.message = message
        self.exceptionMessage = exceptionMessage
        self.exceptionStackTrace = exceptionStackTrace


class FatalReply (WarningReply): pass



MESSAGE = 'message'
EXCEPTION_MESSAGE = 'exceptionMessage'
EXCEPTION_STACK_TRACE = 'exceptionStackTrace'


def decodeReply (data):

    doc = parseString(data)
    de = doc.documentElement

    if de.tagName == OK_REPLY:
        return OkReply()

    message = de.getAttribute(MESSAGE)
    exceptionMessage = de.getAttribute(EXCEPTION_MESSAGE)
    exceptionStackTrace = de.getAttribute(EXCEPTION_STACK_TRACE)

    if de.tagName == WARNING_REPLY:
        return WarningReply(message, exceptionMessage, exceptionStackTrace)

    if de.tagName == FATAL_REPLY:
        return FatalReply(message, exceptionMessage, exceptionStackTrace)

    raise ValueError, "Cannot handle replies by name '%s'" % de.tagName


#
# WORKITEM CONSUMER


class WorkitemConsumer (Service):

    def use (self, workitem):
        self.linfo('\n -- workitem --\n')
        self.linfo(workitem)
        self.linfo('\n')

#
# SOCKET LISTENER


DEFAULT_PORT = 7010

HOST = 'host'
PORT = 'port'
CONSUMER = 'consumer'


def removeAdLine (xmlString):
    if xmlString[0:1] == '<': return xmlString
    i = xmlString.find('\n')
    #print 'removed >%s<' % xmlString[:i]
    #print "what's left : --|%s|--" % xmlString[i+1:]
    return xmlString[i+1:]


class ListeningThread (Thread):

    listener = None
    port = DEFAULT_PORT
    workitemConsumer = None

    mainsocks = []
    readsocks = {}
    writesocks = []


    def run (self):

        portsock = socket(AF_INET, SOCK_STREAM)
        portsock.bind(('', self.port))
        portsock.listen(5)
        self.mainsocks.append(portsock)

        self.readsocks[portsock] = ''

        #self.listener.ldebug("starting to listen on port %i" % self.port)

        while 1:
            readables, writeables, exceptions = \
                select(self.readsocks.keys(), self.writesocks, [])
            for sockobj in readables:
                if sockobj in self.mainsocks:
                    newsock, address = sockobj.accept()
                    self.readsocks[newsock] = ('', 0)
                        # (workitem, retry)
                else:
                    workitem, retry = self.readsocks[sockobj]
                    data = sockobj.recv(7 * 1024)
                    if not data:
                        sockobj.close()
                        del self.readsocks[sockobj]
                        #if retry == MAX_RETRY:
                        #    sleep(0.001)
                        #    self.readsocks[sockobj] = (workitem, retry+1)
                        #elif retry > MAX_RETRY:
                        #    sockobj.close()
                        #    del self.readsocks[sockobj]
                        #else:
                        #    self.readsocks[sockobj] = (workitem, retry-1)
                    else:
                        workitem += data

                        #self.listener.ldebug\
                        #    ('incoming data : \n\n %s\n' % str(data))

                        if RE_END_OF_ITEM.search(workitem, 1):
                            sockobj.send(self.okReply())
                            sockobj.close()
                            del self.readsocks[sockobj]
                            workitem = removeAdLine(workitem)
                            self.workitemConsumer.use(codec.decode(workitem))
                        else:
                            self.readsocks[sockobj] = (workitem, retry)


    def okReply (self):

        return self.reply(OK_REPLY, None)


    def reply (self, replyName, message):

        doc = Document()
        e = doc.createElement(replyName)

        if message != None:
            te = doc.createTextNode(message)
            e.appendChild(te)

        doc.appendChild(e)

        reply = doc.toprettyxml(indent='  ', newl='\n', encoding=getEncoding())

        self.listener.ldebug('\nreplying :\n\n %s' % reply)

        return reply


class SocketListener (Service):

    listeningThread = None

    def init (self, name, context, params):
        Service.init(self, name, context, params)

        self.listeningThread = ListeningThread()

        self.listeningThread.listener = self

        sPort = params.get(PORT)
        if sPort != None: self.listeningThread.port = int(sPort)
        self.linfo("listening on port %i" % self.listeningThread.port)

        consumerName = params.get(CONSUMER)
        if consumerName != None:
            self.listeningThread.workitemConsumer = context.get(consumerName)
            self.linfo("set workitemConsumer to '%s'" % consumerName)
        else:
            raise ServiceException, "Param '%s' is mandatory for service '%s'" % (CONSUMER, self.__class__)
        
        self.listeningThread.start()
        
        self.linfo("started to listen")


#
# SOCKET DISPATCHER

XMLCODER_AD = 'xmlCoder'

class SocketDispatcher (Service):

    host = 'localhost'
    port = 7010

    def init (self, name, context, params):
        Service.init(self, name, context, params)

        if params.has_key(HOST): self.host = params[HOST]
        if params.has_key(PORT): self.port = int(params[PORT])


    def dispatch (self, workitem):

        sckt = socket(AF_INET, SOCK_STREAM)
        sckt.connect((self.host, self.port))

        data = codec.encode(workitem)

        length = len(data)

        data = \
            XMLCODER_AD + ' ' + str(length) + '\n\n' + data

        # seems better for python 2.3.x
        #
        sckt.sendall(data)

        # used to work with python 2.1 and 2.2
        #
        #for line in data.splitlines():
        #    sckt.sendall(line)

        #self.linfo('dispatched :\n\n%s\n' % data)
        
        reply = decodeReply(sckt.recv(8192))

        if not isinstance(reply, OkReply):
            raise ValueError, "error %s" % reply