// Consumer_Router.cpp,v 4.10 2004/01/05 22:57:06 shuston Exp

#include "ace/os_include/os_assert.h"
#include "ace/OS_NS_stdio.h"
#include "ace/OS_NS_string.h"
#include "Consumer_Router.h"
#include "Options.h"

ACE_RCSID(Event_Server, Consumer_Router, "Consumer_Router.cpp,v 4.10 2004/01/05 22:57:06 shuston Exp")

Consumer_Router::Consumer_Router (Peer_Router_Context *prc)
  : Peer_Router (prc)
{
  this->context ()->duplicate ();
}

// Initialize the Router. 

int
Consumer_Router::open (void *)
{
  if (this->is_writer ())
    {
      // Set the <Peer_Router_Context> to point back to us so that if
      // any Consumer's "accidentally" send us data we'll be able to
      // handle it by passing it down the stream.
      this->context ()->peer_router (this);

      // Increment the reference count.
      this->context ()->duplicate ();

      // Make this an active object to handle the error cases in a
      // separate thread.  This is mostly just for illustration, i.e.,
      // it's probably overkill to use a thread for this!
      return this->activate (Options::instance ()->t_flags ());
    }
  else // if (this->is_reader ())

    // Nothing to do since this side is primarily used to transmit to
    // Consumers, rather than receive.
    return 0; 
}

int
Consumer_Router::close (u_long)
{
  ACE_DEBUG ((LM_DEBUG, "(%t) closing Consumer_Router %s\n",
	      this->is_reader () ? "reader" : "writer"));

  if (this->is_writer ())
    // Inform the thread to shut down.
    this->msg_queue ()->deactivate ();

  // Both writer and reader call <release>, so the context knows when
  // to clean itself up.
  this->context ()->release ();
  return 0;
}

// Handle incoming messages in a separate thread.

int 
Consumer_Router::svc (void)
{
  assert (this->is_writer ());

  ACE_DEBUG ((LM_DEBUG, 
	      "(%t) starting svc in Consumer_Router\n"));

  for (ACE_Message_Block *mb = 0;
       this->getq (mb) >= 0;
       )
    {
      ACE_DEBUG ((LM_DEBUG, 
		  "(%t) warning: Consumer_Router is "
                  "forwarding a message to Supplier_Router\n"));

      // Pass this message down to the next Module's writer Task.
      if (this->put_next (mb) == -1)
	ACE_ERROR_RETURN ((LM_ERROR, 
			   "(%t) send_peers failed in Consumer_Router\n"),
			  -1);
    }

  ACE_DEBUG ((LM_DEBUG, 
	      "(%t) stopping svc in Consumer_Router\n"));
  return 0;
  // Note the implicit ACE_OS::thr_exit() via destructor.
}

// Send a <Message_Block> to the supplier(s). 

int 
Consumer_Router::put (ACE_Message_Block *mb, 
		      ACE_Time_Value *)
{
  // Perform the necessary control operations before passing the
  // message down the stream.

  if (mb->msg_type () == ACE_Message_Block::MB_IOCTL)
    {
      this->control (mb);
      return this->put_next (mb);
    }

  // If we're the reader then we're responsible for broadcasting
  // messages to Consumers.

  else if (this->is_reader ())
    {
      if (this->context ()->send_peers (mb) == -1)
	ACE_ERROR_RETURN ((LM_ERROR, 
			   "(%t) send_peers failed in Consumer_Router\n"),
			  -1);
      else 
	return 0;
    }
  else // if (this->is_writer ())

    // Queue up the message to processed by <Consumer_Router::svc>
    // Since we don't expect to be getting many of these messages, we
    // queue them up and run them in a separate thread to avoid taxing
    // the main thread.
    return this->putq (mb);
}

// Return information about the <Consumer_Router>.

int
Consumer_Router::info (char **strp, size_t length) const
{
  char buf[BUFSIZ];
  ACE_INET_Addr  addr;
  const char *mod_name = this->name ();
  
  if (this->context ()->acceptor ().get_local_addr (addr) == -1)
    return -1;
  
  ACE_OS::sprintf (buf,
                   "%s\t %d/%s %s (%s)\n",
		   mod_name,
                   addr.get_port_number (),
                   "tcp",
		   "# consumer router",
                   this->is_reader () ? "reader" : "writer");
  if (*strp == 0 && (*strp = ACE_OS::strdup (mod_name)) == 0)
    return -1;
  else
    ACE_OS::strncpy (*strp, mod_name, length);

  return ACE_OS::strlen (mod_name);
}