src/python/osrf/stack.py

   1 # -----------------------------------------------------------------------
   2 # Copyright (C) 2007  Georgia Public Library Service
   3 # Bill Erickson <billserickson@gmail.com>
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; either version 2
   8 # of the License, or (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 # -----------------------------------------------------------------------
  15
  16 import time
  17 import osrf.json, osrf.log, osrf.ex, osrf.ses, osrf.const, osrf.app
  18
  19 def push(net_msg):
  20
  21     ses = osrf.ses.Session.find_or_create(net_msg.thread)
  22     ses.set_remote_id(net_msg.sender)
  23     if not ses.service:
  24         ses.service = osrf.app.Application.name
  25
  26     omessages = osrf.json.to_object(net_msg.body)
  27
  28     osrf.log.log_internal("stack.push(): received %d messages" % len(omessages))
  29
  30     # Pass each bundled opensrf message to the message handler
  31     start = time.time()
  32     for msg in omessages:
  33         handle_message(ses, msg)
  34     duration = time.time() - start
  35
  36     if isinstance(ses, osrf.ses.ServerSession):
  37         osrf.log.log_info("Message processing duration %f" % duration)
  38
  39     return ses
  40
  41 def handle_message(session, message):
  42
  43     osrf.log.log_internal("handle_message(): processing message of "
  44         "type %s" % message.type())
  45
  46     if isinstance(session, osrf.ses.ClientSession):
  47         handle_client(session, message)
  48     else:
  49         handle_server(session, message)
  50
  51
  52 def handle_client(session, message):
  53
  54     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_RESULT:
  55         session.push_response_queue(message)
  56         return
  57
  58     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_STATUS:
  59
  60         status_code = int(message.payload().statusCode())
  61         status_text = message.payload().status()
  62         osrf.log.log_internal("handle_message(): processing STATUS, "
  63             "status_code =  %d" % status_code)
  64
  65         if status_code == osrf.const.OSRF_STATUS_COMPLETE:
  66             # The server has informed us that this request is complete
  67             req = session.find_request(message.threadTrace())
  68             if req:
  69                 osrf.log.log_internal("marking request as complete: %d" % req.rid)
  70                 req.set_complete()
  71             return
  72
  73         if status_code == osrf.const.OSRF_STATUS_OK:
  74             # We have connected successfully
  75             osrf.log.log_debug("Successfully connected to " + session.service)
  76             session.state = OSRF_APP_SESSION_CONNECTED
  77             return
  78
  79         if status_code == osrf.const.OSRF_STATUS_CONTINUE:
  80             # server is telling us to reset our wait timeout and keep waiting for a response
  81             session.reset_request_timeout(message.threadTrace())
  82             return
  83
  84         if status_code == osrf.const.OSRF_STATUS_TIMEOUT:
  85             osrf.log.log_debug("The server did not receive a request from us in time...")
  86             session.state = OSRF_APP_SESSION_DISCONNECTED
  87             return
  88
  89         if status_code == osrf.const.OSRF_STATUS_NOTFOUND:
  90             osrf.log.log_error("Requested method was not found on the server: %s" % status_text)
  91             session.state = OSRF_APP_SESSION_DISCONNECTED
  92             raise osrf.ex.OSRFServiceException(status_text)
  93
  94         if status_code == osrf.const.OSRF_STATUS_INTERNALSERVERERROR:
  95             raise osrf.ex.OSRFServiceException("Server error %d : %s" % (status_code, status_text))
  96
  97         raise osrf.ex.OSRFProtocolException("Unknown message status: %d" % status_code)
  98
  99
 100 def handle_server(session, message):
 101
 102     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_REQUEST:
 103         osrf.log.log_debug("server received REQUEST from %s" % session.remote_id)
 104         osrf.app.Application.handle_request(session, message)
 105         return
 106
 107     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_CONNECT:
 108         osrf.log.log_debug("server received CONNECT from %s" % session.remote_id)
 109         session.state = osrf.const.OSRF_APP_SESSION_CONNECTED
 110         session.send_connect_ok(message.threadTrace())
 111         return
 112
 113     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_DISCONNECT:
 114         osrf.log.log_debug("server received DISCONNECT from %s" % session.remote_id)
 115         session.state = osrf.const.OSRF_APP_SESSION_DISCONNECTED
 116         return
 117
 118     if message.type() == osrf.const.OSRF_MESSAGE_TYPE_STATUS:
 119         # Should never get here
 120         osrf.log.log_warn("server received STATUS from %s" % session.remote_id)
 121         return
 122
 123