[omniORB] Summary: Using omniNotify with omniOrbPy

[Mark Zimmerman]

Greetings:

Having worked through the initial learning curve of integrating
omniNotify with omniOrbPy, I am posting the following example.

I have a Notification channel working with C++ code and I wanted to be
able to inject structured events into the channel from a python
script.  That is all the attached script does, but if anyone wants to
do Notification stuff from python, this example may help break through
the initial barrier of understanding.

Many thanks to Duncan Grisby and Robert E. Gruber for their assistance
in working through this.

Step 1:

You need to generate the python stub code from the idl code. To do
this, create a Stub directory and execute the following script from
within that directory. Then, append the Stub directory path to your
PYTHONPATH.

----------------------------------------------------------------------
#!/usr/bin/env python

import os

idlfiles = ["TimeBase", "CosTime", "CosEventComm", "CosEventChannelAdmin",
"CosTypedEventComm", "CosTypedEventChannelAdmin", "CosTimerEvent",
"CosNotification", "CosNotifyComm", "CosNotifyFilter",
"CosNotifyChannelAdmin", "CosTypedNotifyComm", "CosTypedNotifyChannelAdmin",
"AttNotifyChannelAdmin"]

# d is the top-level idl directory from an omniORB release
d = "/your/path/to/omni/idl"

for x in idlfiles:
    y = x + ".idl"
    print x
    cmd = "omniidl -bpython -I%s -I%s/COS -DNOLONGLONG %s/COS/%s" % (d, d, d, y)
    print cmd
    os.system(cmd)
----------------------------------------------------------------------

Step 2:

Get your Notification channel running if you don't have one already.

Step 3:

The following script works for me. To keep the example short there is
no error handling but the comments essentially say everything I know
right now.

----------------------------------------------------------------------
#!/usr/bin/env python

import os
import sys
import time
from omniORB import CORBA
import CosNaming
import CosNotification
import CosNotifyComm
import CosNotifyComm__POA
import CosNotifyChannelAdmin

#+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
#
class Pusher (CosNotifyComm__POA.StructuredPushSupplier):

#-----------------------------------------------------------------------------
#
   def __init__ (self):         # Constructor

#.............................................................................
#     Initialize the ORB.
#
      orb = CORBA.ORB_init(sys.argv, CORBA.ORB_ID)

#.............................................................................
#     Resolve the naming service.
#
      ns = orb.resolve_initial_references("NameService")
      rootCon = ns._narrow(CosNaming.NamingContext)

#.............................................................................
#     Resolve the event channel.
#
      name = [CosNaming.NameComponent("EventChannel","EventChannel")]
      echannel_ref = rootCon.resolve(name)
      channel = echannel_ref._narrow(CosNotifyChannelAdmin.EventChannel)

#.............................................................................
#     Get the Notification Channel Admin object.
#
      sadmin, suppid = channel.new_for_suppliers(CosNotifyChannelAdmin.AND_OP)

#.............................................................................
#     Get the push consumer proxy and connect to it.
#
      cons, prxID = sadmin.obtain_notification_push_consumer (
                       CosNotifyChannelAdmin.STRUCTURED_EVENT)
      self._pcons = cons._narrow (
                       CosNotifyChannelAdmin.StructuredProxyPushConsumer)
      self._pcons.connect_structured_push_supplier(self._this())

#-----------------------------------------------------------------------------
#
   def Speak (                  # Interface to send a message
      self,                     # object pointer
      severity,                 # integer indicating an error level
      whosez,                   # string identifying the sender
      whatsay):                 # string containing the message

#.............................................................................
#     Build the event type and the fixed header. The event name can be
#     used in constraint expressions by the receiver but to the sender,
#     any arbitrary string will work.
#
      etype = CosNotification.EventType ("SPW", "Log")
      ename = "Strange_women_lying_in_ponds_distributing_swords"
      fixed_header = CosNotification.FixedEventHeader(etype, ename)

#.............................................................................
#     Build the variable header. Mine is empty, but if you need one it is
#     built exactly like the filterable data array, below.
#
      variable_header = [ ]

#.............................................................................
#     Create the header.
#
      header = CosNotification.EventHeader (fixed_header, variable_header)

#.............................................................................
#     Gather some information about the caller. (Irrelevant to the example)
#
      hostname = os.uname()[1]
      username = os.environ['USER']
      pid      = os.getpid()
      ppid     = os.getppid()
      curtime  = int(time.time())

#.............................................................................
#     Build the filterable data array.
#
      filterable_data = [
         CosNotification.Property ( "Severity",
                                    CORBA.Any(CORBA._tc_ulong,  severity)),
         CosNotification.Property ( "HostName",
                                    CORBA.Any(CORBA._tc_string, hostname)),
         CosNotification.Property ( "UserName",
                                    CORBA.Any(CORBA._tc_string, username)),
         CosNotification.Property ( "PID",
                                    CORBA.Any(CORBA._tc_ulong,  pid)),
         CosNotification.Property ( "PPID",
                                    CORBA.Any(CORBA._tc_ulong,  ppid)),
         CosNotification.Property ( "ProgramName",
                                    CORBA.Any(CORBA._tc_string, whosez)),
         CosNotification.Property ( "Time",
                                    CORBA.Any(CORBA._tc_ulong,  curtime)),
      ]

#.............................................................................
#     Convert the message to type Any.
#
      remainder_of_body = CORBA.Any(CORBA._tc_string, whatsay)

#.............................................................................
#     Construct the event and send it on its way.
#
      event = CosNotification.StructuredEvent (
                 header, filterable_data, remainder_of_body)
      self._pcons.push_structured_event (event)

#+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
#
if __name__ == '__main__':

   jack = Pusher()
   jack.Speak (0, "PythonScript", "Hello, sailor")
   jack.Speak (1, "PythonScript", "Hello, sailor")
   jack.Speak (2, "PythonScript", "Hello, sailor")

----------------------------------------------------------------------