TAO/examples/Simulator/DOVEBrowser/PushConsumer.java

   1 //
   2 // ============================================================================
   3 //
   4 // = FILENAME
   5 //    PushConsumer.java
   6 //
   7 // = AUTHOR
   8 //    Michael Kircher (mk1@cs.wustl.edu)
   9 //
  10 // = DESCRIPTION
  11 //    This is a Push Consumer which takes the data field of the
  12 //    event and updates with it a Data Handler.
  13 //
  14 // ============================================================================
  15
  16
  17
  18 // The Consumer has to implement the Skeleton Consumer
  19
  20 public class PushConsumer extends RtecEventComm.PushConsumerPOA
  21 {
  22
  23   public static final int ACE_ES_EVENT_ANY = 0;
  24   public static final int ACE_ES_EVENT_SHUTDOWN = 1;
  25   public static final int ACE_ES_EVENT_ACT = 2;
  26   public static final int ACE_ES_EVENT_NOTIFICATION = 3;
  27   public static final int ACE_ES_EVENT_TIMEOUT = 4;
  28   public static final int ACE_ES_EVENT_INTERVAL_TIMEOUT = 5;
  29   public static final int ACE_ES_EVENT_DEADLINE_TIMEOUT = 6;
  30   public static final int ACE_ES_GLOBAL_DESIGNATOR = 7;
  31   public static final int ACE_ES_CONJUNCTION_DESIGNATOR = 8;
  32   public static final int ACE_ES_DISJUNCTION_DESIGNATOR = 9;
  33   public static final int ACE_ES_EVENT_UNDEFINED = 16;
  34   public static final int TOTAL_MESSAGES = 30;
  35
  36   // Store the number of received events
  37   private int total_received_ = 0;
  38   private org.omg.CORBA.ORB orb_;
  39   private org.omg.PortableServer.POA poa_;
  40   private MTDataHandlerAdapter dataHandlerAdapter_;
  41   private int rt_info_;
  42   private RtecEventChannelAdmin.EventChannel channel_admin_;
  43   private RtecEventChannelAdmin.ConsumerAdmin consumer_admin_;
  44   private RtecEventChannelAdmin.ProxyPushSupplier suppliers_;
  45
  46   public PushConsumer (org.omg.CORBA.ORB orb,
  47                        org.omg.PortableServer.POA poa,
  48                        DataHandler dataHandler,
  49                        boolean use_queueing)
  50     {
  51       orb_ = orb;
  52       poa_ = poa;
  53       dataHandlerAdapter_ =
  54         new MTDataHandlerAdapter (dataHandler, use_queueing);
  55       if (use_queueing)
  56         {
  57           dataHandlerAdapter_.start ();
  58         }
  59     }
  60
  61
  62   public void push (RtecEventComm.Event[] events)
  63     {
  64       if (total_received_ < 5)
  65         System.out.println ("Demo Consumer: Received an event set! ->Number: "
  66                             + total_received_);
  67       else if (total_received_ == 5)
  68         System.out.println ("Demo Consumer: Everything is fine. " +
  69                             "Going to be mute.");
  70
  71       if (events.length == 0)
  72         {
  73           System.err.println ("No events");
  74         }
  75       else
  76         {
  77           total_received_++;
  78           dataHandlerAdapter_.push (events);
  79         }
  80     }
  81
  82   public void disconnect_push_consumer()
  83     {
  84       System.out.println ("Demo Consumer: Have to disconnect!");
  85     }
  86
  87   public void open_consumer (RtecEventChannelAdmin.EventChannel event_channel_,
  88                              RtecScheduler.Scheduler scheduler_,
  89                              String name)
  90     {
  91       try {
  92
  93         // Define Real-time information
  94
  95         rt_info_ = scheduler_.create (name);
  96
  97         scheduler_.set (rt_info_,
  98                         RtecScheduler.Criticality_t.VERY_LOW_CRITICALITY,
  99                         0L,
 100                         0L,
 101                         0L,
 102                         2500000,  // period
 103                         RtecScheduler.Importance_t.VERY_LOW_IMPORTANCE,
 104                         0L,
 105                         1,
 106                         RtecScheduler.Info_Type_t.OPERATION);
 107
 108
 109         // Register for Notification and Shutdown events
 110
 111         byte payload[] = new byte[1];
 112         payload[0] = 0;
 113         RtecEventComm.Event notification_event_ = new RtecEventComm.Event ();
 114         notification_event_.header =
 115           new RtecEventComm.EventHeader (ACE_ES_EVENT_NOTIFICATION,
 116                                          0, 1, 0, 0, 0);
 117
 118         notification_event_.data =
 119           new RtecEventData (0, payload, orb_.create_any());
 120
 121         RtecEventChannelAdmin.Dependency dependencies_[] = new RtecEventChannelAdmin.Dependency[1];
 122         dependencies_[0] = new RtecEventChannelAdmin.Dependency (notification_event_, rt_info_);
 123
 124
 125         // @@ Carlos please help me to set the right boolean value
 126         RtecEventChannelAdmin.ConsumerQOS qos = new RtecEventChannelAdmin.ConsumerQOS (dependencies_, false);
 127
 128
 129         // The channel administrator is the event channel we got from the invocation
 130         // of this routine
 131
 132         channel_admin_ = event_channel_;
 133
 134         // Connect as a consumer
 135
 136         consumer_admin_ = channel_admin_.for_consumers ();
 137
 138         // Obtain a reference to the proxy push supplier
 139
 140         suppliers_ = consumer_admin_.obtain_push_supplier ();
 141
 142         org.omg.CORBA.Object objref = poa_.servant_to_reference (this);
 143         RtecEventComm.PushConsumer consumer_ref =
 144           RtecEventComm.PushConsumerHelper.narrow (objref);
 145         suppliers_.connect_push_consumer (consumer_ref, qos);
 146
 147         System.out.println ("Registered the consumer successfully.");
 148
 149
 150   }
 151   catch (RtecEventChannelAdmin.TypeError e)
 152         {
 153           System.err.println ("Demo_Consumer.open_consumer: RtecEventChannelAdmin.TypeError");
 154           System.err.println (e);
 155         }
 156   catch (RtecEventChannelAdmin.AlreadyConnected e)
 157         {
 158           System.err.println ("Demo_Consumer.open_consumer: RtecEventChannelAdmin.AlreadyConnected");
 159           System.err.println (e);
 160         }
 161   catch (RtecScheduler.UNKNOWN_TASK e)
 162         {
 163           System.err.println ("Demo_Consumer.open_consumer: Unknown task");
 164           System.err.println (e);
 165         }
 166   catch (RtecScheduler.DUPLICATE_NAME e)
 167         {
 168           System.err.println ("Demo_Consumer.open_consumer: Duplicate names");
 169           System.err.println (e);
 170         }
 171   catch (RtecScheduler.INTERNAL e)
 172         {
 173           System.err.println ("Demo_Consumer.open_consumer: internal scheduler error");
 174           System.err.println (e);
 175         }
 176   catch (RtecScheduler.SYNCHRONIZATION_FAILURE e)
 177         {
 178           System.err.println ("Demo_Consumer.open_consumer: scheduler synchronization failure");
 179           System.err.println (e);
 180         }
 181   catch (org.omg.PortableServer.POAPackage.ServantNotActive e)
 182         {
 183           System.err.println ("Demo_Consumer.open_consumer: org.omg.PortableServer.POAPackage.ServantNotActive");
 184           System.err.println (e);
 185         }
 186   catch (org.omg.PortableServer.POAPackage.WrongPolicy e)
 187         {
 188           System.err.println ("Demo_Consumer.open_consumer: org.omg.PortableServer.POAPackage.WrongPolicy");
 189           System.err.println (e);
 190         }
 191   catch(org.omg.CORBA.SystemException e)
 192         {
 193           System.err.println(e);
 194         }
 195     }
 196 }