ACE/examples/APG/Streams/Answerer.cpp

   1 /**
   2  * Streams Listing 01
   3  *
   4  * An answering machine based on a one-way ACE_Stream
   5  */
   6
   7 #include "ace/OS_NS_string.h"
   8 #include "ace/Stream.h"
   9 #include "ace/Message_Block.h"
  10 #include "ace/FILE_IO.h"
  11
  12 #include "MessageInfo.h"
  13 #include "Message.h"
  14 #include "BasicTask.h"
  15 #include "EndTask.h"
  16 #include "Util.h"
  17 #include "RecordingDevice.h"
  18
  19 // Listing 21 code/ch18
  20 class AnswerIncomingCall : public BasicTask
  21 {
  22 protected:
  23   virtual int process (Message *message)
  24   {
  25     ACE_TRACE ("AnswerIncomingCall::process()");
  26
  27     if (message->recorder ()->answer_call () < 0)
  28       ACE_ERROR_RETURN ((LM_ERROR,
  29                          ACE_TEXT ("%p\n"),
  30                          ACE_TEXT ("AnswerIncomingCall")),
  31                         -1);
  32       return 0;
  33   }
  34 };
  35 // Listing 21
  36
  37 // Listing 22 code/ch18
  38 class GetCallerId : public BasicTask
  39 {
  40 protected:
  41   virtual int process (Message *message)
  42   {
  43     ACE_TRACE ("GetCallerId::process()");
  44
  45     CallerId *id;
  46     id = message->recorder ()->retrieve_callerId ();
  47     if (!id)
  48       ACE_ERROR_RETURN ((LM_ERROR,
  49                          ACE_TEXT ("%p\n"),
  50                          ACE_TEXT ("GetCallerId")),
  51                         -1);
  52
  53     message->caller_id (id);
  54     return 0;
  55   }
  56 };
  57 // Listing 22
  58
  59 // Listing 23 code/ch18
  60 class PlayOutgoingMessage : public BasicTask
  61 {
  62 protected:
  63   virtual int process (Message *message)
  64   {
  65     ACE_TRACE ("PlayOutgoingMessage::process()");
  66
  67     ACE_FILE_Addr outgoing_message =
  68       this->get_outgoing_message (message);
  69
  70     int pmrv =
  71       message->recorder ()->play_message (outgoing_message);
  72     if (pmrv < 0)
  73       ACE_ERROR_RETURN ((LM_ERROR,
  74                          ACE_TEXT ("%p\n"),
  75                          ACE_TEXT ("PlayOutgoingMessage")),
  76                         -1);
  77     return 0;
  78   }
  79
  80   ACE_FILE_Addr get_outgoing_message (Message *)
  81   {
  82     // Exclude 23
  83     return ACE_FILE_Addr (ACE_TEXT ("/tmp/outgoing_message"));
  84     // Exclude 23
  85   }
  86 };
  87 // Listing 23
  88
  89 // Listing 24 code/ch18
  90 class RecordIncomingMessage : public BasicTask
  91 {
  92 protected:
  93   virtual int process (Message *message)
  94   {
  95     ACE_TRACE ("RecordIncomingMessage::process()");
  96
  97     ACE_FILE_Addr incoming_message =
  98       this->get_incoming_message_queue ();
  99
 100     MessageType *type =
 101       message->recorder ()->record_message (incoming_message);
 102     if (!type)
 103       ACE_ERROR_RETURN ((LM_ERROR,
 104                          ACE_TEXT ("%p\n"),
 105                          ACE_TEXT ("RecordIncomingMessage")),
 106                         -1);
 107     message->incoming_message (incoming_message, type);
 108     return 0;
 109   }
 110
 111   ACE_FILE_Addr get_incoming_message_queue ()
 112   {
 113     // Exclude 24
 114     return ACE_FILE_Addr (ACE_TEXT ("/tmp/incoming_message"));
 115     // Exclude 24
 116   }
 117 };
 118 // Listing 24
 119
 120 // Listing 25 code/ch18
 121 class ReleaseDevice : public BasicTask
 122 {
 123 protected:
 124   virtual int process (Message *message)
 125   {
 126     ACE_TRACE ("ReleaseDevice::process()");
 127     message->recorder ()->release ();
 128     return 0;
 129   }
 130 };
 131 // Listing 25
 132
 133 // Listing 26 code/ch18
 134 class EncodeMessage : public BasicTask
 135 {
 136 protected:
 137   virtual int process (Message *message)
 138   {
 139     ACE_TRACE ("EncodeMessage::process()");
 140
 141     ACE_FILE_Addr &incoming = message->addr ();
 142     ACE_FILE_Addr addr = this->get_message_destination (message);
 143
 144     if (message->is_text ())
 145       Util::convert_to_unicode (incoming, addr);
 146     else if (message->is_audio ())
 147       Util::convert_to_mp3 (incoming, addr);
 148     else if (message->is_video ())
 149       Util::convert_to_mpeg (incoming, addr);
 150
 151     message->addr (addr);
 152     return 0;
 153   }
 154
 155   ACE_FILE_Addr get_message_destination (Message *)
 156   {
 157     // Exclude 26
 158     return ACE_FILE_Addr (ACE_TEXT ("/tmp/encoded_message"));
 159     // Exclude 26
 160   }
 161 };
 162 // Listing 26
 163
 164 // Listing 27 code/ch18
 165 class SaveMetaData : public BasicTask
 166 {
 167 protected:
 168   virtual int process (Message *message)
 169   {
 170     ACE_TRACE ("SaveMetaData::process()");
 171
 172     ACE_TString path (message->addr ().get_path_name ());
 173     path += ACE_TEXT (".xml");
 174
 175     ACE_FILE_Connector connector;
 176     ACE_FILE_IO file;
 177     ACE_FILE_Addr addr (path.c_str ());
 178     if (connector.connect (file, addr) == -1)
 179       ACE_ERROR_RETURN ((LM_ERROR,
 180                          ACE_TEXT ("%p\n"),
 181                          ACE_TEXT ("create meta-data file")),
 182                         0);
 183
 184     file.truncate (0);
 185     this->write (file, "<Message>\n");
 186     // ...
 187     this->write (file, "</Message>\n");
 188     file.close ();
 189     return 0;
 190   }
 191
 192 private:
 193   //FUZZ: disable check_for_lack_ACE_OS
 194   int write (ACE_FILE_IO &file, const char *str)
 195   {
 196   //FUZZ: enable check_for_lack_ACE_OS
 197     return file.send (str, ACE_OS::strlen (str));
 198   }
 199 };
 200 // Listing 27
 201
 202 // Listing 28 code/ch18
 203 class NotifySomeone : public BasicTask
 204 {
 205 protected:
 206   virtual int process (Message *message)
 207   {
 208     ACE_TRACE ("NotifySomeone::process()");
 209
 210     // Format an email to tell someone about the
 211     // newly received message.
 212     // ...
 213
 214     // Display message information in the logfile
 215     ACE_DEBUG ((LM_INFO,
 216                 ACE_TEXT ("New message from %s ")
 217                 ACE_TEXT ("received and stored at %s\n"),
 218                 message->caller_id ()->string (),
 219                 message->addr ().get_path_name ()));
 220     return 0;
 221   }
 222 };
 223 // Listing 28
 224
 225 // Listing 10 code/ch18
 226 class RecordingStream : public ACE_Stream<ACE_MT_SYNCH>
 227 {
 228 public:
 229   typedef ACE_Stream<ACE_MT_SYNCH> inherited;
 230   typedef ACE_Module<ACE_MT_SYNCH> Module;
 231
 232   RecordingStream () : inherited()
 233   { }
 234 // Listing 10
 235
 236   //FUZZ: disable check_for_lack_ACE_OS
 237   // Listing 1000 code/ch18
 238   virtual int open (void *arg,
 239                     Module *head = 0, Module *tail = 0)
 240   {
 241   //FUZZ: enable check_for_lack_ACE_OS
 242     if (tail == 0)
 243       ACE_NEW_RETURN (tail,
 244                       Module (ACE_TEXT ("End Module"), new TheEndTask ()),
 245                       -1);
 246     this->inherited::open (arg, head, tail);
 247     // Listing 1000
 248
 249     // Listing 1001 code/ch18
 250     Module *answerIncomingCallModule;
 251     ACE_NEW_RETURN (answerIncomingCallModule,
 252                     Module (ACE_TEXT ("Answer Incoming Call"),
 253                             new AnswerIncomingCall ()),
 254                     -1);
 255
 256     // Listing 11 code/ch18
 257     Module *getCallerIdModule;
 258     ACE_NEW_RETURN (getCallerIdModule,
 259                     Module (ACE_TEXT ("Get Caller ID"), new GetCallerId ()),
 260                     -1);
 261     // Listing 11
 262
 263     Module *playOGMModule;
 264     ACE_NEW_RETURN (playOGMModule,
 265                     Module (ACE_TEXT ("Play Outgoing Message"),
 266                             new PlayOutgoingMessage ()),
 267                     -1);
 268
 269     Module *recordModule;
 270     ACE_NEW_RETURN (recordModule,
 271                     Module (ACE_TEXT ("Record Incoming Message"),
 272                             new RecordIncomingMessage ()),
 273                     -1);
 274
 275     Module *releaseModule;
 276     ACE_NEW_RETURN (releaseModule,
 277                     Module (ACE_TEXT ("Release Device"),
 278                             new ReleaseDevice ()),
 279                     -1);
 280
 281     Module *conversionModule;
 282     ACE_NEW_RETURN (conversionModule,
 283                     Module (ACE_TEXT ("Encode Message"),
 284                             new EncodeMessage ()),
 285                     -1);
 286
 287     Module *saveMetaDataModule;
 288     ACE_NEW_RETURN (saveMetaDataModule,
 289                     Module (ACE_TEXT ("Save Meta-Data"),
 290                             new SaveMetaData ()),
 291                     -1);
 292
 293     Module *notificationModule;
 294     ACE_NEW_RETURN (notificationModule,
 295                     Module (ACE_TEXT ("Notify Someone"),
 296                             new NotifySomeone ()),
 297                     -1);
 298     // Listing 1001
 299
 300     // Listing 12 code/ch18
 301     if (this->push (notificationModule) == -1)
 302       ACE_ERROR_RETURN ((LM_ERROR,
 303                          ACE_TEXT ("Failed to push %p\n"),
 304                          ACE_TEXT ("notificationModule")),
 305                         -1);
 306     if (this->push (saveMetaDataModule) == -1)
 307       ACE_ERROR_RETURN ((LM_ERROR,
 308                          ACE_TEXT ("Failed to push %p\n"),
 309                          ACE_TEXT ("saveMetaDataModule")),
 310                         -1);
 311     if (this->push (conversionModule) == -1)
 312       ACE_ERROR_RETURN ((LM_ERROR,
 313                          ACE_TEXT ("Failed to push %p\n"),
 314                          ACE_TEXT ("conversionModule")),
 315                         -1);
 316     if (this->push (releaseModule) == -1)
 317       ACE_ERROR_RETURN ((LM_ERROR,
 318                          ACE_TEXT ("Failed to push %p\n"),
 319                          ACE_TEXT ("releaseModule")),
 320                         -1);
 321     if (this->push (recordModule) == -1)
 322       ACE_ERROR_RETURN ((LM_ERROR,
 323                          ACE_TEXT ("Failed to push %p\n"),
 324                          ACE_TEXT ("recordModule")),
 325                         -1);
 326     if (this->push (playOGMModule) == -1)
 327       ACE_ERROR_RETURN ((LM_ERROR,
 328                          ACE_TEXT ("Failed to push %p\n"),
 329                          ACE_TEXT ("playOGMModule")),
 330                         -1);
 331     if (this->push (getCallerIdModule) == -1)
 332       ACE_ERROR_RETURN ((LM_ERROR,
 333                          ACE_TEXT ("Failed to push %p\n"),
 334                          ACE_TEXT ("getCallerIdModule")),
 335                         -1);
 336     if (this->push (answerIncomingCallModule) == -1)
 337       ACE_ERROR_RETURN ((LM_ERROR,
 338                          ACE_TEXT ("Failed to push %p\n")
 339                          ACE_TEXT ("answerIncomingCallModule")),
 340                         -1);
 341     // Listing 12
 342
 343     return 0;
 344   }
 345
 346   // Listing 13 code/ch18
 347   int record (RecordingDevice *recorder)
 348   {
 349     ACE_Message_Block * mb = 0;
 350     ACE_NEW_RETURN (mb, ACE_Message_Block (sizeof(Message)), -1);
 351
 352     Message *message = (Message *)mb->wr_ptr ();
 353     mb->wr_ptr (sizeof(Message));
 354
 355     ACE_DEBUG ((LM_DEBUG,
 356                 ACE_TEXT ("RecordingStream::record() - ")
 357                 ACE_TEXT ("message->recorder(recorder)\n")));
 358     message->recorder (recorder);
 359
 360     int rval = this->put (mb);
 361     ACE_DEBUG ((LM_DEBUG,
 362                 ACE_TEXT ("RecordingStream::record() - ")
 363                 ACE_TEXT ("this->put() returns %d\n"),
 364                 rval));
 365     return rval;
 366   }
 367   // Listing 13
 368 };
 369
 370
 371 // Listing 1 code/ch18
 372 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 373 {
 374   RecordingDevice *recorder =
 375     RecordingDeviceFactory::instantiate (argc, argv);
 376   // Listing 1
 377
 378   // Listing 2 code/ch18
 379   RecordingStream *recording_stream;
 380   ACE_NEW_RETURN (recording_stream, RecordingStream, -1);
 381
 382   if (recording_stream->open (0) < 0)
 383     ACE_ERROR_RETURN ((LM_ERROR,
 384                        ACE_TEXT ("%p\n"),
 385                        ACE_TEXT ("RecordingStream->open()")),
 386                       0);
 387   // Listing 2
 388
 389   // Listing 3 code/ch18
 390   for (;;)
 391     {
 392       ACE_DEBUG ((LM_INFO,
 393                   ACE_TEXT ("Waiting for incoming message\n")));
 394       RecordingDevice *activeRecorder =
 395         recorder->wait_for_activity ();
 396
 397       ACE_DEBUG ((LM_INFO,
 398                   ACE_TEXT ("Initiating recording process\n")));
 399
 400       recording_stream->record (activeRecorder);
 401     }
 402   // Listing 3
 403
 404   ACE_NOTREACHED (return 0;)
 405 }