// stream.cpp,v 1.4 1999/09/22 03:13:50 jcej Exp

// stream.cxx
//
// Tutorial regarding a way to use ACE_Stream.
//
// written by bob mcwhirter (bob@netwrench.com)
//
//

#include "Task.h"
#include "EndTask.h"
// This is our specialized ACE_Task.

#include <ace/Module.h>
#include <ace/Stream.h>
#include <ace/streams.h>
// These are the neccessary ACE headers.


typedef ACE_Module<ACE_MT_SYNCH> Module;
typedef ACE_Stream<ACE_MT_SYNCH> Stream;
// Just to avoid a lot of typing, typedefs
// are generally a good idea.

int main(int argc, char *argv[])
{
  int numberOfMessages = argc > 1 ? ACE_OS::atoi(argv[1]) : 3;
  // unless otherwise specified, just send three messages
  // down the stream.

  Stream theStream;
  // the ACE_Stream itself.

  // Now, we instantiate 4 different Tasks.  These do not
  // need to be all the same class, but they do need to
  // all derrive from the same flavor of ACE_Task.
  //
  // Also, we instantiate a fifth end-cap Task to clean
  // up Message_Blocks as they reach the end.

  Task *taskOne;
  Task *taskTwo;
  Task *taskThree;
  Task *taskFour;
  Task *taskEnd;

  // Out Task's take two arguments: a name, and the number
  // of threads to dedicate to the task.

  taskOne = new Task("Task No. 1", 1);
  taskTwo = new Task("Task No. 2", 3);
  taskThree = new Task("Task No. 3", 7);
  taskFour = new Task("Task No. 4", 1);

  // Our EndTask only takes 1 argument, as it actually
  // doesn't spawn any threads for processing.

  taskEnd = new EndTask("End Task");

  Module *moduleOne;
  Module *moduleTwo;
  Module *moduleThree;
  Module *moduleFour;
  Module *moduleEnd;

  // ACE_Stream accepts ACE_Modules, which are simply a pair of
  // ACE_Tasks.  One is dedicated for writing, while the other
  // is dedicated to reading.  Think of the writing side as
  // downstream, and the reading side as upstream.
  //
  // We're only working with a unidirection Stream today,
  // so we'll only actually install a Task into the write
  // side of the module, effectively downstream.

  moduleOne = new Module("Module No. 1", taskOne);
  moduleTwo = new Module("Module No. 2", taskTwo);
  moduleThree = new Module("Module No. 3", taskThree);
  moduleFour = new Module("Module No. 4", taskFour);
  moduleEnd = new Module("Module End", taskEnd);

  // Now we push the Modules onto the Stream.
  // Pushing adds the module to the head, or
  // otherwise prepends it to whatever modules
  // are already installed.

  // So, you need to push() the modules on -backwards-
  // from our viewpoint.

  if (theStream.push(moduleEnd) == -1) {
           ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), -1);
  }

  if (theStream.push(moduleFour) == -1) {
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), -1);
  }

  // As we push a Module onto the Stream, it gets opened.
  // When a Module open()s, it opens the Tasks that it contains.
  //
  // Since we cannot provide an argument to this embedded
  // call to open(), we supplied specified the number of
  // threads in the constructor of our Tasks.

  if (theStream.push(moduleThree) == -1) {
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), -1);
  }

  if (theStream.push(moduleTwo) == -1) {
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), -1);
  }

  if (theStream.push(moduleOne) == -1) {
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "push"), -1);
  }

  // Now that the Modules are open, the Tasks threads should
  // be launching and entering their svc() loop, so we send
  // some messages down the Stream.

  int sent = 1;

  ACE_Message_Block *message;

  while (sent <= numberOfMessages) {

    // First, create ourselves a Message_Block.
    // see Tutorials 10-13 for more information
    // about Message_Blocks and Message_Queues.

    message = new ACE_Message_Block(128);

    // Now, we grab the write-pointer from the Block,
    // and sprintf() our text into it.

    ACE_OS::sprintf(message->wr_ptr(), "Message No. %d", sent);

    // All we have to do now is drop the Message_Block
    // into the Stream.

    // It is always a good idea to duplicate() a Message_Block
    // when you put it into any Message_Queue, as then
    // you can always be allowed to release() your copy
    // without worry.

    if (theStream.put(message->duplicate(), 0) == -1) {
      ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "put"), -1);
    }

    message->release();
    ++sent;
  }

  // Now that we've sent our Message_Blocks, close down
  // the Stream.
  //
  // The Stream will automagically delete the Modules and
  // the contained Tasks.  We don't have to do that.
  //
  // This call will block (due to the way we've written our
  // Task class) until all Message_Blocks have cleared the
  // entire Stream, and all associated threads have exited.

  theStream.close();

  return 0;
}