% A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR
% CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
% EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
% PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
% PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
% LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
% NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
% SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

\chapter{Communications in \rhexlib}

\section{Introduction}

The RHexLib communications facilities are built with the assumption that the
communications channels are unreliable, and possibly intermittent.  For
example, we do not use the TCP/IP messaging protocols.  TCP/IP provides
guaranteed messaging between processes, but the overhead and uncertainties of
guaranteeing that message delivery in the face of an unreliable communications
medium such as outdoor wireless ethernet networks, pose significant problems
and soures of unpredictable behavior.  Instead of TCP/IP, we have chosen to
base our packages on the UDP (User Datagram Protocol), in which messages are
not sent over ``connections'' but rather simply sent to a destination with no
garauntee of receipt.  We build simple, but robust, mechanisms on top of UDP to
allow some garaunteed message delivery.  Since we have built these mechanisms
ourselves, we can more finely specify and make predictable the behavior of the
message delivery in a possibly unreliable wireless ethernet environment.

The basic communications in RHexLib uses a communications manager. This
communications manager provides interface to various UDP based protocols for
sending messages between processes.  On top of this package we build several
convenience packages to make interprocess communications easier.  The first of
these uses a communications manager to set up and manage a ``database'' which
can be distributed across processes.  This database can be used to propagate
information about the robot state when publishing from the robot to outside
clients, and can be used as the central clearing house for commands when
database changes come from outside clients to the robot.  Another of these can
be used to initiate high bandwidth logging of data, and to reliable retrieve
this logged data across the network.  

An important note: the RHexLib communications libraries make no provisions for
communication between machines with different byte orders.  In fact, you even
have to be careful when sending flat C or C++ structures between processors.
Either the two sides have to be compiled with the same compiler, or you have to
carefully organize the structure to avoid any problems with differing methods
of data alignment.  For example, some compilers will pad a one byte type such
as a unsigned char out to four bytes when preceding a four byte integer.
Others will not.  So a structure which has the elements ``unsigned char''
followed by an ``int'' is much more dangerous than the structure which has the
elements ``int'' followed by an ``unsigned char''.

\section {Message sending}

There are two paradigms for communications in RHexLib:  the mailbox paradigm
and the stream paradigm.  

The mailbox paradigm is the simpler of the two.  Mailbox messages are very
close in spirit to raw UDP packets, i.e., users just dispatch mail to a
mailbox, and there is no guarantee that the mail will ever arrive, or even
arrive in the correct order.  Mailbox messages are not queued, so only the most
recently received mail message is available to be read.  Now, in a reliable
communications environment, it is very unlikely that mail will not be
delievered, and it is extremely unlikely that mail will be received out of
order, but there are no guarantees on either.

The stream paradigm provides some measure of guarantees on delivery.  Users
establish stream sinks as destinations for messages.  Users connect stream
sources to those stream sinks in order to send messages.  Every stream message
that is received by a streak sink is acknowledged, and stream sources will
periodically resend messages if they are not acknowledged.  The stream source
holds a queue of the received messages for processing by the user.  This
protocol is analogous to TCP/IP, but is much simpler, and users have much finer
access to and control over the list of unacknowledged messages left in the
stream sources.  Note that while this protocol can guarantee that a message
sent is received, its simplicity means that in the case of intermittent
communications failures, it is actually very likely that there will be multiple
copies of a message received at a stream sink, and that the messages will be
received out of order.

Although the individual components have their classes defined in a variety of
header files, you will only have to include the file {\tt RHexCom.hh}.

\subsection {Communications Manager}

The CommManager class is at the heart of communications in RHexLib.  All
mailboxes and streams are created using an instance of the CommManager class.
In addition, the creation of the CommManager class initiates a sub-thread which
manages the periodic resending of unacknowledged stream messages.  The
CommManager class also maintains a list of all remote communications managers
which the various stream sources and mailers are contacting.  Another
sub-thread is created which processes incoming UDP messages and sorts them into
the proper mailboxes and stream sinks.

\begin{codesegment}
#include "RHexCom/CommManager.hh"
\end{codesegment}

\begin{classdef}
class CommManager {
public:
  CommManager(unsigned short port, int resend_millisecs=200, int priority=-1);
  ~CommManager();

  com_id_t getID();

  Mailbox* createMailbox(int size, com_id_t mailbox_id, 
                         com_id_t exclusive_client=0);
  void destroyMailbox(Mailbox*);

  Mailer* createMailer(const char* host, unsigned short port,
                       int size, com_id_t mailbox_id,
                       com_id_t exclusive_dest=0);
  void destroyMailer(Mailer*);

  StreamSink* createStreamSink(int size, com_id_t stream_id,
                               int max_buffer, com_id_t exclusive_client=0);
  void destroyStreamSink(StreamSink*);

  StreamSource* createStreamSource(RemoteManager* dest,
                                   int size, com_id_t stream_id,
                                   com_id_t exclusive_dest=0);
  void destroyStreamSource(StreamSource*);

  RemoteManager* lookupRemote(const char* host, unsigned short port);
  RemoteManager* openRemote(const char* host, unsigned short port);

  int sendMailMsg(RemoteManager* dest, com_id_t mailbox_id, Message* msg);
};
\end{classdef}

\begin{prototype}
CommManager(unsigned short port, int resend\_millisecs=200, int priority=-1);
\end{prototype}

This is the CommManager constructor.  {\tt port} is the actual UDP port number
which this communications manager will use to send and receive messages.  The
constructor initiates a thread which has two basic purposes.  The main purpose
is to resend stream messages which have not yet been acknowledged.  A secondary
purpose is to notice that it has not heard from a remote communications manager
in a while, and if this is the case try and ``ping'' that module with a dummy
stream message which will then be acknowledged.  Thus, even if a remove
communications manager never explicity sends any information to this module, we
can maintain a ``heartbeat'' to establish that the remote communications
manager is actually alive and healthy.  The sub-thread runs roughly every {\tt
resend\_millisecs}, and it defaults to 200ms.  If {\tt priority} is negative,
then it indicates a relative POSIX thread priority of {\tt priority} less than
the current for the communication manager sub-threads.  If {\tt priority} is
not negative, it indicates an absolute POSIX thread priority.  {\tt priority}
defaults to one priority less than the current.

\begin{prototype}
~CommManager()
\end{prototype}

The destructor cleans up the CommManager memory and terminates the resend
thread. 

\begin{prototype}
com\_id\_t getID()
\end{prototype}

Every communications has a non-zero identifier.  It should be unique across
all communications managers, but it does simply use the lower 32 bits of the
time in microseconds at the time the CommManager was constructed.  If by some
amazing coincidence, two communications managers do have the same ID, it is not
a fatal problem, it just means that there is a chance that some operations that
are supposed to be ``exclusive'' to one manager will be open to both.

\begin{prototype}
Mailbox* createMailbox(int size, com\_id\_t mailbox\_id, 
                       com\_id\_t exclusive\_client=0)
void destroyMailbox(Mailbox*)
\end{prototype}

Create and destroy mailboxes for receiving mail.  A mailbox must be given a
maximum size in bytes, {\tt size}, and an identifier, {\tt mailbox\_id}.  If
{\tt exclusive\_client} is non-zero, then only messages originating from a
communications manager with identifier {\tt exclusive\_client} are accepted.


\begin{prototype}
Mailer* createMailer(const char* host, unsigned short port,
                     int size, com\_id\_t mailbox\_id,
                     com\_id\_t exclusive\_dest=0)
void destroyMailer(Mailer*)
\end{prototype}

Create and destroy mailers for sending mail to a mailbox.  Mail will be sent to
a mailbox managed by a remote communications manager on the machine {\tt host}
with UDP port {\tt port}.  The destination mailbox will have ID {\tt
mailbox\_id} and must be have a capacity of at least {\tt size} bytes.  If {\tt
exclusive\_dest} is non-zero, then the remote manager of the destination mailbox
must have ID {\tt exclusive\_dest}.  This covers the case where you are sending
mail, and the remote manager dies and restarts and you don't want mail going to
the newly created mailbox without restarting some other protocol.

\begin{prototype}
StreamSink* createStreamSink(int size, com\_id\_t stream\_id,
                             int max\_buffer, com\_id\_t exclusive\_client=0)
void destroyStreamSink(StreamSink*)
\end{prototype}

Create and destroy stream sinks for receiving streamed messages.  The stream ID
will be {\tt stream\_id}.  Any message coming to this stream sink must have a
maximum length of {\tt size} bytes.  If the user ignores this sink, it will
buffer up to {\tt max\_buffer} messages before dropping the oldest messages.
If {\tt exclusive\_client} is non-negative then this sink will only accept
messages from a remote communications manager with ID {\tt exclusive\_client}.

\begin{prototype}
StreamSource* createStreamSource(const char* host, unsigned short port,
                                 int size, com\_id\_t stream\_id,
                                 com\_id\_t exclusive\_dest=0)
void destroyStreamSource(StreamSource*)
\end{prototype}

Create and destroy stream sources for sending streamed messages to remote
stream sinks.  The remote stream sink should have ID {\tt stream\_id} on the
manager running on the maching {\tt host} on UDP port {\tt port}.  The remote
stream sink should have at least {\tt size} bytes of capacity.  If {\tt
exclusive\_dest} is non-zero, then if the destination manager changes ID's from
{\tt exclusive\_dest}, then no more messages will be sent.

\begin{prototype}
RemoteManager* lookupRemote(const char* host, unsigned short port)
\end{prototype}

If any of the mailer's or stream sources have contacted a communications
manager on the machine {\tt host} monitoring UDP port {\tt port}, return a
reference to that remote manager.  Otherwise, return NULL.

\begin{prototype}
RemoteManager* openRemote(const char* host, unsigned short port)
\end{prototype}

If any of the mailer's or stream sources have contacted a communications
manager on the machine {\tt host} monitoring UDP port {\tt port}, return a
reference to that remote manager.  If there is no contact with that remote
manager there, attempt to make contact, and establish a ping relationship with
the remote manager.

\begin{prototype}
int sendMailMsg(RemoteManager* dest, com\_id\_t mailbox\_id, Message* msg)
\end{prototype}

Creating a mailer is sometimes overkill.  This method allows you to send the
message {\tt msg} directly to the mailbox with ID {\tt mailbox\_id} managed by
the remote manager {\tt dest} without having to explicitly create a mailer.
The return value is -1 for an error, or the number of bytes sent.

\subsection {Messages}

All of the communication manager's mailboxes and streams are sending and
receiving messages, i.e., instances of the class Message.  Primarily a message
wraps an array of bytes of a fixed size, and then indicates how much of that
array actually containes data.  The Message class also provides access to
message time stamps, and message sources as well as providing convenience
methods for packaging and unpackaging the message data.

Most of the time you will not be creating or destroying messages, as the
creation and destruction of Message instances will be handled by the higher
level entities.  In general, you should not do it on your own as the mailboxes,
mailers, stream sources, and stream sinks have facilities for efficiently
managing messages to minimize the number of memory allocations and copies
occuring in message processing.

\begin{codesegment}
#include "RHexCom/Message.hh"
\end{codesegment}

\begin{classdef}
class Message {
 public:
  Message(int max_size);
  ~Message();

  unsigned char* getData() const;
  int getSize() const;
  void setSize(int size);
  int getMaxSize() const;
  TimeStamp getTimeStamp() const;
  RemoteManager* getSource() const;

  int getInt() const;
  double getDouble() const;
  bool setInt(int i);
  bool setDouble(double d);
  template <class T> bool setStruct( T* val);
  template <class T> bool getStruct( T* val);
};

\end{classdef}

\begin{prototype}
Message(int max\_size)
\end{prototype}

Create a message with a maximum size of {\tt max\_size}.

\begin{prototype}
~Message()
\end{prototype}

Free up all memory associated with a message.

\begin{prototype}
unsigned char* getData() const
\end{prototype}

Get a pointer to the message's block of data.

\begin{prototype}
int getSize() const
\end{prototype}

Get the number of bytes in the message's block of data.

\begin{prototype}
void setSize(int size)
\end{prototype}

Set the number of bytes in the message's block of data.

\begin{prototype}
int getMaxSize() const
\end{prototype}

Get the maximum size of the message's block of data.

\begin{prototype}
TimeStamp getTimeStamp() const
\end{prototype}

In the case of a received message, this method returns the time of receipt.

\begin{prototype}
RemoteManager* getSource() const
\end{prototype}

In the case of a received message, this method returns a reference to the
message sender.

\begin{prototype}
int getInt() const
\end{prototype}

Convenience method for extracting an integer out of the message block.  If the
size of the message data is less than the size of an integer, the method
returns 0.

\begin{prototype}
double getDouble() const
\end{prototype}

Convenience method for extracting a double precision floating point number out
of the message block.  If the size of the message data is less than the size of
a double, the method returns 0.

\begin{prototype}
bool setInt(int i)
\end{prototype}

Convenience method for setting the message's data to be the integer {\tt i}.
If the maximum size of the message data is less than the size of an integer,
the method returns false and does nothing, otherwise the method sets the
message data and the message size appropriately and returns true.

\begin{prototype}
bool setDouble(double d)
\end{prototype}

Convenience method for setting the message's data to be the double {\tt d}.  If
the maximum size of the message data is less than the size of an integer, the
method returns false and does nothing, otherwise the method sets the message
data and the message size appropriately and returns true.

\begin{prototype}
template <class T> bool getStruct( T* val)
\end{prototype}

Templated convenience method for extracting a structure of type {\tt T} from
the message data.  If the size of the message data is greater than the size of
type {\tt T}, then the appropriate number of bytes are copied from the message
data to the pointer {\tt val} and true is returned.  Otherwise, nothing is done
and false is returned.

\begin{prototype}
template <class T> bool setStruct( T* val)
\end{prototype}

Templated convenience method for copying a structure of type {\tt T} to the
message data.  If the maximum size of the message is greater than the size of
type {\tt T}, then the structure pointed to by {\tt val} is copied into the
message data and the message size is set to the size of the type {\tt T}, and
true is returned.  Otherwise false nothing is done and false is returned.

\subsubsection {Message usage and examples}

The Message class is structured to allow you to manipulate message data in a
variety of different ways, but there are two general styles which can be used.

One style minimizes the amount of data which is copied from place to place, at
the expense of ``extra'' code that you have to generate.  For example, say we
have a message that we want to fill with the following structure,

\begin{codesegment}
        struct TestStruct {
          int a;
          int b;
          bool c;
        };
\end{codesegment}

The most efficient way, from a performance perspective, to set up and send this
message is,

\begin{codesegment}
        TestStruct* ts = (TestStruct*) msg->getData();
        ts->a = 1;
        ts->b = 2;
        ts->c = false;
        msg->setSize(sizeof(TestStruct);
\end{codesegment}

Similarly, if we want to unpack a {\tt TestStruct} instance from a message, we
could do,

\begin{codesegment}
        TestStruct* ts = (TestStruct*) msg->getData();
        if (msg->getSize() < sizeof(testStruct))
          return false;
        a = ts->a;
        b = ts->b;
        c = ts->c;
\end{codesegment}

As a convenience, if you do not mind that there is an extra copying of the
data, you can use the convenience templated methods {\tt getStruct} and {\tt
setStruct}.  For example, to copy the message data to a destination structure,
you could simply do,

\begin{codesegment}
        TestStruct ts;
        if (!msg->getStruct<TestStruct>(&ts))
          return false;
\end{codesegment}

And to copy a source structure to the message data you can do,

\begin{codesegment}
        TestStruct ts;
        if (!msg->setStruct<TestStruct>(&ts))
          return false;
\end{codesegment}

This style is much more ``compact'' than dealing with the message data
directly, but that compactness does come at the price of an extra message copy.

\subsubsection{Message structure caveats}

Remember that messages are ultimately just a vector of bytes.  There are
several things to watch out for when directly using C++ structures to set and
get message data.

First, you must be aware that different compilers can ``align'' structure
elements differently.  For example, some compilers will pad a one byte type
such as a unsigned char out to four bytes when preceding a four byte integer.
Others will not.  So a structure which has the elements ``unsigned char''
followed by an ``int'' is much more dangerous than the structure which has the
elements ``int'' followed by an ``unsigned char''.

Even more importantly, remember that even under the best of circumstances, only
``flat'' structures can be dealt with properly by the message convenience
functions.  A flat structure cannot be a class, and cannot contain pointers of
any kind.  For example, this is a flat structure,

\begin{codesegment}
       struct FlatStruct {
         int a, b;
         float c[24];
       }
\end{codesegment}

But, this is not a flat structure,

\begin{codesegment}
        struct UnflatStruct {
          int a, b;
          float *c;
        }
\end{codesegment}

If you have a structure that is not flat that you want to send, you will have
to manage the packing and unpacking yourself.  The simplest example is the
sending and receiving of a variable length string.

On the sending side, one way to package a string ({\tt char* s}) is this,

\begin{codesegment}
        int len = strlen(s);
        if (len > msg->getMaxSize())  // truncate strings which are too large
          len = msg->getMaxSize()-1;
        memcpy(msg->getData(), s, len);
        msg->getData()[len] = '\0';
        msg->setSize(len+1);
\end{codesegment}

On the receiving side, you simply unpack the string by doing,

\begin{codesegment}
        char* s = (char*) msg->getData();
\end{codesegment}

Packaging more complicated structures, such as multiple strings, or structures
containing variable length vectors of number, has to be done ``by hand,'',
i.e., you must assemble the data vector component by component.  On the plus
side, assembling the data vector by hand this way alleviates the problem of
structure alignment differences between compilers.

\subsection {Mailing}

Mail is the most primitive form of communications in RHexLib.  A send to a
mailbox results in an immediate dispatch of a UDP message.  There is no
guarantee that this message will get through, and no guarantee that later
messages will not be received first.  The receiver mailbox simply has an
indicator of whether new mail has been received or not.  The user can examine
the mailbox's current contents, which reflects only the last mail received:
any older messages are overwritten.

\subsubsection{Mailboxes}

\begin{codesegment}
#include "RHexCom/Mailbox.hh"
\end{codesegment}

\begin{classdef}
class Mailbox {
 public:
  Message* getData();   
  void releaseMsg(Message*);  
  bool newMail() const;
  int getMaxSize() const;
  com_id_t getID() const;

  int getInt();
  double getDouble();

};

\end{classdef}

\begin{prototype}
Message* getData()
\end{prototype}

Get the current data in the mail box.

\begin{prototype}
void releaseMsg(Message* msg)
\end{prototype}

After you are through with the message returned by {\tt getData}, you must
release it back to the mailbox.  If you do this religiously, then you will make
mailbox operations vastly more efficient, reducing the number of memory copies,
allocations, and deallocations to a bare minimum.
  
\begin{prototype}
bool newMail() const
\end{prototype}

Returns true if a mail message has arrived since the last invocation of {\tt
getData}.

\begin{prototype}
int getMaxSize() const
\end{prototype}

Returns the mailbox capacity in bytes.

\begin{prototype}
com\_id\_t getID() const
\end{prototype}

Returns the mailbox identifier.

\begin{prototype}
int getInt()
double getDouble()
\end{prototype}

Convenience functions for if the mailbox just contains integers or double
precision floating point numbers.  Note:  mailboxes are not free, if you have a
large number of associated values it is much better to group them as a single
mailbox containing a structure rather than a group of mailboxes with integers
and doubles, despite the ``ease of use'' of integer and double access.
Grouping into one mailbox saves overhead on the communications manager, as well
as guaranteeing simultenaity of the grouped data.

\subsubsection{Mailers}

\begin{codesegment}
#include <RhexCom/Mailer.hh>
\end{codesegment}

\begin{classdef}

class Mailer {
 public:
  Message* createMsg();
  int sendMsg(Message*);
  void releaseMsg(Message* m);

  // convenience functions
  int sendInt(int i);
  int sendDouble(double d);

  RemoteManager* getDest() const;
  com_id_t getID() const;
  int getMaxSize() const;
};

\end{classdef}

\begin{prototype}
Message* createMsg()
\end{prototype}

Create the message to be sent.  It will be set with the maximum capacity of the
mailbox as specified by {\tt CommManager::createMailer}.

\begin{prototype}
int sendMsg(Message*)
\end{prototype}

Send the message to the remote mailbox.  If successful, the message will be
released and the value 0 will be returned.  If the mailbox has been
invalidated for any reason, then the message will not be released, and the
return value will be -1.

\begin{prototype}
void releaseMsg(Message* m)
\end{prototype}

If you create a message but do not send it, or {\tt sendMessage} returns -1,
then you must use this method to release the message.

\begin{prototype}
RemoteManager* getDest() const
\end{prototype}

Return the reference to the communication manager which handles the destination
mailbox. 

\begin{prototype}
com\_id\_t getID() const
\end{prototype}

Returns the mailbox ID.

\begin{prototype}
int getMaxSize() const
\end{prototype}

Returns the maximum capacity of the mailbox.

\begin{prototype}
int sendInt(int i)
int sendDouble(double d)
\end{prototype}

Convenience routines for sending integers and doubles.  Again:  use with
caution, as it is usually better to bundle related values together in a
structure rather than having many individual integer and double mailboxes.

\subsubsection {Mailer example}

This example sends mail at a regular interval.

\begin{codesegment}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "RHexCom.hh"

#define MAILBOX_ID 100
struct MailStruct {
  int a, b;
  unsigned char c;
};

int main(int argc, char** argv)
{
  if (argc != 4) {
    fprintf(stderr, "Usage: mailer <port> <dest_machine> <dest_port>\n");
    exit(-1);
  }
  
  // create the communications manager
  unsigned short port = atoi(argv[1]);
  CommManager* mgr = new CommManager(port);
  
  // create the mailer to send to a remote communications manager
  Mailer* mailer = mgr->createMailer(argv[2], atoi(argv[3]),
                                     sizeof(MailStruct), MAILBOX_ID);
  for (int i=1;;i++) {
    Message* msg = mailer->createMsg(); // create the message

    // fill the message using the "direct" method
    MailStruct* ms = (MailStruct*) msg->getData();
    ms->a = i;
    ms->b = i*2;
    ms->c = i % 2;

    // send the message
    if (mailer->sendMsg(msg) < 0) {
      // this is probably paranoid, but illustrated the proper way to handle
      // a send error.  Send errors are more likely to happen with "exclusive"
      // mailboxes, which will become invalid if the destination manager dies 
      // and restarts.
      printf("Mailbox invalidated\n");
      mailer->releaseMsg(msg);
      break;
    }
    printf("Sent %d\n", i);
    usleep(1000000);
  }

  delete mgr;   // This is all we need for cleaning up
}

\end{codesegment}

\subsubsection {Mailbox example}

This example polls a mailbox at a regular interval.

\begin{codesegment}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "RHexCom.hh"

// note these must be consistent from the previous example
#define MAILBOX_ID 100
struct MailStruct {
  int a, b;
  unsigned char c;
};

int main(int argc, char** argv)
{
  if (argc != 2) {
    fprintf(stderr, "Usage: mailbox <port>\n");
    exit(-1);
  }
  
  // create the communications manager
  unsigned short port = atoi(argv[1]);
  CommManager* mgr = new CommManager(port);

  // create the mailbox
  Mailbox* box = mgr->createMailbox(sizeof(MailStruct), MAILBOX_ID);
  
  while (1) {
    bool new_data = box->newMail(); // check if there is new mail

    // poll the mail box
    Message* msg = box->getData();
    MailStruct* dest= (MailStruct*) msg->getData();
    printf("Received %d %d %d %d\n", (int) new_data,
           dest->a, dest->b, dest->c);

    // release the message back to the mailbox
    box->releaseMsg(msg);

    usleep(500000);
  }

  delete mgr;   // This is all we need for cleaning up
}

\end{codesegment}

\subsection {Streams}

Message streams allow users to create stream sources which send messages with
guarantees on their delivery while still maintaining fine grained control over
the delivery attempts.  Unlike mailboxes, stream sinks (the destination for
stream messages) maintain a queue of incoming messages rather than simply the
last message.

\subsubsection{Stream Sources}

\begin{codesegment}
#include <RhexCom/StreamSource.hh>
\end{codesegment}

\begin{classdef}

class StreamSource {
 public:
  Message* createMsg();
  com_id_t sendMsg(Message*);
  void releaseMsg(Message* m);

  com_id_t sendInt(int i);
  com_id_t sendDouble(double d);

  RemoteManager* getDest() const;
  com_id_t getID() const;
  int getMaxSize() const;

  int ackBlock(com_id_t msg_id, int timeout_ms=-1);
  void flush(com_id_t msg_id=0); 
  int numToSend();
  Message* retreivePendingMsg(com_id_t msg_id);
  Message* popPendingMsg();  
  Message* popOldestMsg();  
};

\end{classdef}

\begin{prototype}
Message* createMsg()
\end{prototype}

Create the message to be sent.  It will be set with the maximum capacity of the
stream as specified by {\tt CommManager::createStreamSource}.

\begin{prototype}
com\_id\_t sendMsg(Message*)
\end{prototype}

Send the message to the remote stream sink.  The sub-thread will continue to
resend the message until it is acknowledged.  Upon acknowledgement, the message
will be released.  The return value is the outgoing message ID, which can be
used to retrieve this message from the pending acknowledgement queue.

\begin{prototype}
void releaseMsg(Message* m)
\end{prototype}

If you create a message but do not send it then you must use this method to
release the message.

\begin{prototype}
com\_id\_t sendInt(int i)
com\_id\_t sendDouble(double d)
\end{prototype}

Convenience routines for sending integers and doubles.  Since streams are even
more expensive than mailboxes, use with caution, as it is usually better to
bundle related values together in a structure rather than having many
individual integer and double streams.


\begin{prototype}
RemoteManager* getDest() const
\end{prototype}

Return the reference to the communication manager which handles the remote
stream sink.

\begin{prototype}
com\_id\_t getID() const
\end{prototype}

Returns the remote stream sink ID.

\begin{prototype}
int getMaxSize() const
\end{prototype}

Returns the maximum capacity of a stream message in this stream.

\begin{prototype}
int ackBlock(com\_id\_t msg\_id, int timeout\_ms=-1)
\end{prototype}

This routine blocks until the sent message with identifier {\tt msg\_id} is no
longer on the queue of messages awaiting acknowledgement.  If {\tt timeout\_ms}
is negative, the routine will block forever, otherwise the method will block at
most {\tt timeout\_ms} milliseconds.


\begin{prototype}
void flush(com\_id\_t msg\_id=0)
\end{prototype}

Remove from the queue of messages awaiting acknowledgement any message with
identifier msg\_id.  If msg\_id is 0, remove all the messages from the queue.
All removed messages are released in the process.
 
\begin{prototype}
int numToSend()
\end{prototype}

Return the size of the queue of messages awaiting acknowledgement.

\begin{prototype}
Message* retreivePendingMsg(com\_id\_t msg\_id)
\end{prototype}

If there is a message waiting to be acknowledged with identifier {\tt msg\_id},
remove it from the pending acknowledgment queue and return it, otherwise,
return NULL.

\begin{prototype}
Message* popPendingMsg()
\end{prototype}

If the pending acknowledgment queue is non-empty, pop the most recent
message put on the queue off and return it to the user.
  
\begin{prototype}
Message* popOldestMsg()
\end{prototype}

If the pending acknowledgment queue is non-empty, pop the oldest
message put on the queue off and return it to the user.

\subsubsection{Stream Sinks}

\begin{codesegment}
#include <RhexCom/StreamSink.hh>
\end{codesegment}

\begin{classdef}

class StreamSink {
 public:
  Message* waitData(int msecs_timeout); 
  Message* pollData();  
  void releaseMsg(Message*); 

  int numToReceive();
  void setBufferSize(int b);
  int getBufferSize() const;
  int getMaxSize() const;
  com_id_t getID() const;
  void flush();
};

\end{classdef}

\begin{prototype}
Message* waitData(int msecs\_timeout)
\end{prototype}

Wait for a message to arrive for up to msecs\_timeout milliseconds and return
it.  If no message arrives in that time, return NULL.
 
\begin{prototype}
Message* pollData()
\end{prototype}

Check if a new message has arrived.  If so, return it, if not return NULL.
  
\begin{prototype}
void releaseMsg(Message*)
\end{prototype}

Once you are finished with the message generated by either a {\tt waitData} or
a {\tt pollData} you must release it with this method.

\begin{prototype}
int getMaxSize() const
\end{prototype}

Get the maximum size of the messages in the stream

\begin{prototype}
int numToReceive()
\end{prototype}

Return the number of messages waiting to be received.

\begin{prototype}
void setBufferSize(int b)
\end{prototype}

Set the maximum number of messages that can be in the queue waiting to be
processed.  If this maximum queue size is going to be exceeded, the stream sink
will throw away the oldest pending message.  If {\tt b} is -1, there is no
limit.  This is the initial value if you never call this method.

\begin{prototype}
int getBufferSize() const
\end{prototype}

Get the maximum number of messages that can be in the queue waiting to be
processed.  A negative number indicates that there is no limit.

\begin{prototype}
void flush()
\end{prototype}

Flush all messages in the queue waiting to be processed.  All of the messages
are released.

\begin{prototype}
com\_id\_t getID() const
\end{prototype}

Return the stream ID.

\subsubsection{Stream source example}

This code sends stream messages at a regular interval.  Just as an
illustration, it will give up if the queue of unacknowledged messages gets too
long.

\begin{codesegment}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "RHexCom.hh"

#define STREAM_ID 100
struct StreamStruct {
  int a, b;
  unsigned char c;
};

int main(int argc, char** argv)
{
  if (argc != 4) {
    fprintf(stderr,"Usage: stream_source <port> <dest_machine> <dest_port>\n");
    exit(-1);
  }

  // create communications manager
  unsigned short port = atoi(argv[1]);
  CommManager* mgr = new CommManager(port);

  // create the stream source
  StreamSource* source =
    mgr->createStreamSource(argv[2], atoi(argv[3]),
                            sizeof(StreamStruct), STREAM_ID);

  // send the message periodically
  StreamStruct ss;
  for (int i=1;;i++) {
    //  note we are using the less efficient
    // but more compact method for stuffing messages
    ss.a = i; ss.b = 2*i; ss.c = i % 2;
    Message* msg = source->createMsg();
    msg->setStruct<StreamStruct>(&ss);

    // send the message
    source->sendMsg(msg);
    printf("Sent %d (queuing %d for acknowledgement)\n", i,
           source->numToSend());

    // if too long with no acknowledgement, quit
    if (source->numToSend() > 10)
      break;
    usleep(500000);
  }

  delete mgr;  // clean up
}

\end{codesegment}

\subsubsection{Stream sink example}

This example sits on a stream and waits for new messages to come in.

\begin{codesegment}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "RHexCom.hh"

// the following must be consistent with the previous example
#define STREAM_ID 100
struct StreamStruct {
  int a, b;
  unsigned char c;
};

int main(int argc, char** argv)
{
  if (argc != 4) {
    fprintf(stderr,"Usage: stream_sink <port>\n");
    exit(-1);
  }

  // create communications manager
  unsigned short port = atoi(argv[1]);
  CommManager* mgr = new CommManager(port);

  // create the stream sing
  StreamSink* sink = mgr->createStreamSink(sizeof(int), STREAM_ID, 10);
  Message* msg;
  StreamStruct ss;
  while (1) {
    bool received = false;
    // while there are messages to be processed
    while ((msg = sink->pollData())) {
      // use the more "compact" means for processing data structure
      msg->getStruct<StreamStruct>(&ss);
      printf("Received %d %d\n", msg->getSize(), ss.a);
      sink->releaseMsg(msg);       // release the message back to the sink
      received = true;
    }
    if (received)
      printf("\n");
    usleep(1000000);
  }

  delete mgr; // clean everything up
}

\end{codesegment}

\subsection {Remote managers}

When you create a mailer or a stream source you pass in a machine and a UDP
port number.  Internally, this creates a reference to the remote communications
manager which is maintained by an instance of the class RemoteManager.
In addition, for every instance of RemoteManager has a record of the last time
communications was received from that remote communications manager and what
that remote manager's identifier currently is.

Your communications manager does not merely maintain a record of the last
contact time in a RemoteManager, but for every RemoteManager it maintains, if
the contact time is older than a certain threshold, one of the communication
manager's sub-threads will attempt to ``ping'' the remote communications
manager to elicit a response.  Thus, a RemoteManager can act as the
``heartbeat'' counter even for a communications manager which from the user's
point of view should only be receiving information.

Normally, RemoteManager's will be instantiated in the course of creating
mailers and stream sources, but you can also use the {\tt openRemote} method of
the CommManager class to directly add a host and port to the list of
RemoteManagers to be pinged.

\begin{codesegment}
#include "RHexCom/CommManager.hh"
\end{codesegment}

\begin{classdef}
class RemoteManager {
 public:
  struct in_addr getAddress() const;
  unsigned short getPort() const;
  com_id_t getID() const;
  TimeStamp getTimeStamp() const;
  int lastContactTime(); 
};

\end{classdef}

\begin{prototype}
struct in\_addr getAddress() const
\end{prototype}

Returns the network address of the remote communication manager's machine.

\begin{prototype}
unsigned short getPort() const
\end{prototype}

Returns the UDP port monitored by the remote communication manager.

\begin{prototype}
com\_id\_t getID() const
\end{prototype}

Returns the last known identifier for the remote communication manager.

\begin{prototype}
TimeStamp getTimeStamp() const
\end{prototype}

Returns the timestamp of the last message received from the remote
communication manager.

\begin{prototype}
int lastContactTime()
\end{prototype}

Returns the number of milliseconds since the last message was received from the
remote communications manager.

\subsubsection {Ping example}

This example deliberately misuses the RemoteManager facility to set up a ping
of a remote communications manager.

\begin{codesegment}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "RHexCom.hh"

int main(int argc, char** argv)
{
  if (argc != 4) {
    fprintf(stderr, "Usage: ping_test <port> <dest_machine> <dest_port>\n");
    exit(-1);
  }
  
  unsigned short port = atoi(argv[1]);
  CommManager* mgr = new CommManager(port);

  RemoteManager* remote = mgr->openRemote(argv[2], atoi(argv[3]));
  remote->setTimeStamp(CommManager::now()); // kick off the ping

  while (1) {
    printf("Last contacted %d\n", remote->lastContactTime());
    sleep(1);
  }

  delete mgr;
}

\end{codesegment}

\section {Database operations}

One of the major applications of the communications manager is to publish robot
information and parameterize robot operations.  To facilitate this, we have
built a database abstraction on top of the communication facilities that allows
a user to create a central database of named entries, where remote clients can
subscribe to the entries or register to change the entries by name, without
having to know anything about the details of the communications manager. 

The database is owned by a central manager, typically residing on the robot.
Database clients, such as graphical user interfaces, connect to the database
manager and register to either subscribe to database entries (i.e., when the
entry values change on the robot, the new values are propagated to the clients)
or to publish to database entries (i.e., when the entry value changes on the
client, the new values are propagated to the manager).  Clients cannot register
to both publish and subscribe to the same entry.

There are two methods by which information can be propagated between database
clients and managers.  In immediate mode, as soon as a change is made, that
change is propagated.  Changes made in immediate mode can be throttled to save
network bandwidth, for example, a limit can be set to send changes in a
database entries at most every n milliseconds.  In periodic mode, the value of
the database entry is propagated over the network at a regular interval.

Changes in database entries are propagated by both database managers and
clients through a subthread which contains a list of database entry monitors.
When a database entry is marked as having an external ``client'', a monitor is
added to the list.  The subthread iterates through the list at a regular
interval to check all of the monitors.  Periodic monitors fire at regular
intervals and send the current value to the recipient.  Immediate monitors
simply check to see if there is a throttled change pending to send.

Users can also add callbacks to entries which invoke a C++ method when a
database entry is changed.  These callbacks are implemented in the simplest
possible manner, which means they are very inefficient, and really should not
be used.  When you add a callback to an entry, a monitor gets added to the
monitor list.  Whenever the monitor list checks this monitor, it will perform a
comparison of the old entry value and the new entry value.  If they are
different, then the callback is marked to be executed by a update function that
the user of both data base managers and clients must call.  Note that this
means that callbacks will not be invoked immediately, add significant
computational burden to the monitor thread, and require the user to
periodically invoke a special method in the ``main'' thread of control.  It is
my opinion that callbacks should not be used at all, and that you should design
your code accordingly.  If there is a great hue and cry for them, these
drawbacks can be alleviated somewhat, but it will take more restructuring of
code than you think to make them efficient and effective.

Although the individual components have their classes defined in a variety of
header files, you will only have to include the file {\tt Database.hh}.

\subsection {Database Entries}

The core of the database are the named database entries.  First, each entry has
a name by which it is known in the database.  What the entry contains is
essentially a block of bytes: You can consider a database entry as ``wrapping''
an instance of a communiations manager message, with all of the caveats about
byte order (must be identical across processes), structure alignment (must be
compatible between compilers), and structure components (there can be no
pointers unless you pack the data yourself).

In addition, each entry has a ``description.''  The purpose of a description is
to give some measure of run-time type checking for the data that goes into and
out of the entry as remote clients subscribe to it.  Thus, the description is
usually the name of the type which resides in the entry data.  For example, if
a database entry simply contains an integer, its description would be ``int.''
The entry converts the description into a single integer key for comparison.

Note that database entries can be even more expensive than communication
mailboxes and streams, and thus it is imperative that the database not be
simply a bunch of named integers and doubles.  Data that is meant to be used
together should be grouped together into structures as much as possible.

\begin{codesegment}
#include "RHexCom/DBEntry.hh"
\end{codesegment}

\begin{classdef}
class DBEntry {
public:
  Message* get();
  void release(Message*);

  Message* createData();
  bool set(Message*);

  // convenience methods
  int getInt();
  double getDouble();
  bool setInt(int i);
  bool setDouble(double d);
  template <class T> bool setStruct( T* val);
  template <class T> bool getStruct( T* val);

  const char* getName() const;
  unsigned int getDescriptionKey() const;
  int getMaxSize();
};
\end{classdef}

\begin{prototype}
Message* get()
\end{prototype}

Get the message which contains the entry's data.

\begin{prototype}
void release(Message* msg)
\end{prototype}

Release the message gotten by the {\tt get} method (or, if necessary by the
{\tt createData} method, although that will be very uncommon).

\begin{prototype}
Message* createData()
\end{prototype}

Get a pointer to a message to use to set the entry data.

\begin{prototype}
bool set(Message* msg)
\end{prototype}

Set the entry data using the data in {\tt msg};

\begin{prototype}
int getInt()
double getDouble()
bool setInt(int i)
bool setDouble(double d)
\end{prototype}

Convenience methods for accessing data entries which just contain integers and
double precision floating point numbers.

\begin{prototype}
template <class T> bool setStruct( T* val)
template <class T> bool getStruct( T* val)
\end{prototype}

Convenience templates for copying the entry data to and from {\tt val}, an
instance of type {\tt T}.  Note that this has many of the same advantages and
disadvantages as the similar Message template functions:  it is a compact
operation, but involves wholesale copying which may not be necessary.

\begin{prototype}
const char* getName() const
\end{prototype}

Get the name of the entry as it is entered in the database.

\begin{prototype}
unsigned int getDescriptionKey() const
\end{prototype}

Get the result of running a hash function on the entry's description string.

\begin{prototype}
int getMaxSize()
\end{prototype}

Get the maximum size of the entry data.

\subsection {Common database operations}

Both the database manager and the database client classes have a common
superclass which contains methods common to both, such as the creation and
management of the monitor thread or the management of callbacks.

\begin{codesegment}
#include "RHexCom/DBBase.hh"
\end{codesegment}

\begin{classdef}

class DBBase {
public:
  DBBase(CommManager* mgr, int fundamental_rate, int priority=-1);
  ~DBBase();

  void addCallback(DBEntry* entry, DBCallback* cb);
  void updateCallbacks();

  int getRate() const;
  CommManager* getManager();
};
\end{classdef}

\begin{prototype}
DBBase(CommManager* mgr, int fundamental\_rate, int priority=-1)
\end{prototype}

Both database manager and database clients must use a communications manager,
{\tt mgr}, and create an entry monitoring subthread which runs roughly every
{\tt fundamental\_rate} milliseconds.  If {\tt priority} is negative, then this
sub-thread runs at a relative priority of the current thread priority plus {\tt
priority}.  If {\tt priority} is positive, then the sub-thread will run with
the absolute priority {\tt priority}.

\begin{prototype}
~DBBase()
\end{prototype}

Cleans out the database memory and shuts down the monitoring thread.

\begin{prototype}
void addCallback(DBEntry* entry, DBCallback* cb)
\end{prototype}

Add a callback {\tt cb} to be invoked sometime after a change to the database
entry {\tt entry}.  {\tt cb} is a subclass of the abstract class {\tt
DBCallback}.  Details of this class can be found in {\tt
RHexCom/DBCallback.hh}, and there is also a macro, {\tt DB\_CALLBACK} to help
you use callbacks found in {\tt RHexCom/DBBase.hh}.  I will document this no
more, as you shouldn't use them.

\begin{prototype}
void updateCallbacks()
\end{prototype}

If, through some lack of imagination or sheer stubborness, you do use
callbacks, then your main thread of operation must invoke this method
periodically.  The callbacks are finally invoked in this method.


\begin{prototype}
int getRate() const
\end{prototype}

Returns the number of milliseconds which is the fundamental period of the
monitoring sub-thread.

\begin{prototype}
CommManager* getManager()
\end{prototype}

Get the communications manager passed in at the creation of the class.

\subsection {Database Manager}

There is one database manager which ``owns'' all of the database entries.
Clients can access and even add entries remotely , but the entries are still
truly maintained by the database manager.

As well as creating a monitor thread (through it's superclass, {\tt DBBase}), a
database manager also creates a protocol thread, which is watching a stream
sink for messages from database clients about access to the manager's database.

\begin{codesegment}
#include <RHexCom/DBManager.hh>
\end{codesegment}

\begin{classdef}
class DBManager : public DBBase {
public:
  DBManager(CommManager* mgr, com_id_t id, int fundamental_rate,
            int priority=-1);
  ~DBManager();

  DBEntry* enter(const char* name, int max_size,
                 const char* description, bool local_write_only=false);

  DBEntry* lookup(const char* name, int max_size,
                  const char* description) const;

  // convenience functions
  DBEntry* enterInt(const char* name, bool local_write_only=false);
  DBEntry* enterDouble(const char* name, bool local_write_only=false);
  DBEntry* lookupInt(const char* name);
  DBEntry* lookupDouble(const char* name);

  void enable();
  void disable();

  DBEntry* ENTER(name, type);
  DBEntry* LOOKUP(name, type);
};

\end{classdef}

\begin{prototype}
DBManager(CommManager* mgr, com\_id\_t id, int fundamental\_rate, int priority=-1)
\end{prototype}

Create a database manager.  {\tt mgr}, {\tt fundametnal\_rate}, and {\tt
priority} are passed to the superclass.  When the protocol thread is kicked off
by the {\tt enable} method, it will use the same priority as the monitor thread
and will create a stream sink with the ID {\tt id}.

\begin{prototype}
~DBManager()
\end{prototype}

Shut down the threads properlay and clear out all database memory.


\begin{prototype}
DBEntry* enter(const char* name, int max\_size, const char* description, bool local\_write\_only=false)
\end{prototype}

Create an entry.  If an entry by that name already exists and is
compatible with the maximum size and description entered here, just return that
pre-existing entry.  If the existing entry is incompatible with {\tt max\_size}
and {\tt description}, return NULL.  If no entry exists by that name, create
the appropriate entry and return it.  If local\_write\_only is true, only local
writing to the entry is allowed, i.e., remote clients will not be allowed to
write to this entry.

\begin{prototype}
DBEntry* lookup(const char* name, int max\_size, const char* description) const
\end{prototype}

Lookup an entry.  If the entry already exists and is compatible with {\tt
max\_size} and {\tt description}, just return it. If the entry is incompatible
return NULL.


\begin{prototype}
DBEntry* enterInt(const char* name, bool local\_write\_only=false)
DBEntry* enterDouble(const char* name, bool local\_write\_only=false)
DBEntry* lookupInt(const char* name)
DBEntry* lookupDouble(const char* name)
\end{prototype}

These are convenience methods to create and lookup entries which contain
integers and double precision floating point numbers.

\begin{prototype}
void enable()
\end{prototype}

Launch the protocol thread.  Before this method is invoked, all incoming
messages from remote clients will be ignored.

\begin{prototype}
void disable()
\end{prototype}

Shutdown the protocol thread.

\begin{prototype}
DBEntry* ENTER(name, type)
DBEntry* LOOKUP(name, type)
\end{prototype}

These look like methods, but they are actually convenience macros for entering
and looking up entries named {\tt name} which will contain data of type {\tt
type}.  The size and description of the entry will be extracted using macro
magic from the type name {\tt type}.

\subsubsection{Database manager example}

{\tt Needs to be done: Take a look at RHexLib/base/db\_test.cc}

\subsection {Database Clients}

Users access the central database managed by a database manager by using the
database clients.  These clients send messages to the database manager's stream
sink, and performs all of the necessary handshaking to setup database
communications.  Unlike the database manager operations, which return
DBEntry's, the database client operations return a subclass, RemoteDBEntry
which allows the user to detect when the remote database manager has gone down
and come back up, and allows the user to quickly reregister the entry with the
new database manager.

Note there is one serious lack in the database interface: currently while
remote clients can register to subscribe to database entries, there is no way
for them to cancel a subscription.  The database manager is smart enough to see
that an client has stopped and restarted, and will ``cancel'' all of the
client's previous subscriptions (thus the necessity of the RemoteDBEntry
class), so this is not a fatal flaw, but it is still a flaw.

\begin{codesegment}
#include <RHexCom/DBClient.hh>
\end{codesegment}

\begin{classdef}
class RemoteDBEntry : public DBEntry {
public:
  bool valid() const;
  bool reestablish(int timeout);
\end{classdef}

\begin{prototype}
bool valid() const
\end{prototype}

Returns true when the entry is still valid, i.e., we are connected to the same
database manager which we originally contacted to access or create the entry.

\begin{prototype}
bool reestablish(int timeout)
\end{prototype}

Attempts to reestablish the entry with the ``new'' database with the exact same
flags and parameters.  Returns true if successful within {\tt timeout}
milliseconds, and false otherwise.

\begin{classdef}
class DBClient : public DBBase {
public:
  DBClient(CommManager* mgr,
           const char* host, int port, int id, int fundamental_rate,
           int priority=-1);

  ~DBClient();

  RemoteDBEntry* access(bool periodic, const char* name, int max_size,
                        const char* description,
                        unsigned int access_flags, int rate, int timeout=500);

  RemoteDBEntry* create(bool periodic, const char* name, int max_size,
                        const char* description,
                        unsigned int access_flags, int rate, int timeout=500);

  RemoteDBEntry* accessInt(bool periodic, const char* name,
                           unsigned int access_flags,
                           int rate, int timeout=500);
  RemoteDBEntry* accessDouble(bool periodic, const char* name,
                              unsigned int access_flags,
                              int rate, int timeout=500);
  RemoteDBEntry* createInt(bool periodic, const char* name,
                           unsigned int access_flags,
                           int rate, int timeout=500);
  RemoteDBEntry* createDouble(bool periodic, const char* name,
                              unsigned int access_flags,
                              int rate, int timeout=500);

  RemoteDBEntry* ACCESS(periodic, name, type, access_flags, rate, timeout);
  RemoteDBEntry* CREATE(periodic, name, type, access_flags, rate, timeout);
};
\end{classdef}

\begin{prototype}
DBClient(CommManager* mgr, const char* host, int port, int id, int
fundamental\_rate, int priority=-1)
\end{prototype}

Create a database client which attempts to contact a database manager on the
machine {\tt host} monitoring the stream sink {\tt id} on the UDP port {\tt
port}.  {\tt mgr}, {\tt fundamental\_rate}, and {\tt priority} are passed into
the superclass constructor.

\begin{prototype}
~DBClient()
\end{prototype}

Clean up the client.

\begin{prototype}
RemoteDBEntry* access(bool periodic, const char* name, int max\_size, const char* description, unsigned int access\_flags, int rate, int timeout=500)
\end{prototype}

Attach to an entry on the database manager named {\tt name}.  Return 0 if the
entry exists on the manager, -1 if not.  {\tt access\_flags} is the OR'd value
of the following flags,

\begin{itemize}
\begin{item}
{\tt DB\_PUBLISH}.  If there, we will be writing this value, if not, setting
the entry's value will always fail.
\end{item}

\begin{item}
{\tt DB\_SUBSCRIBE}. If there, we will be reading this value, if not, getting
the entry's value will always fail. 
\end{item}

\begin{item}
{\tt DB\_EXCLUSIVE}. If possible, then no other remote client will be able to write this database element 
\end{item}
\end{itemize}

If {\tt periodic} is true, then when {\tt DB\_PUBLISH} is true, the value of
this entry will be sent to the manager approximately every {\tt rate}
milliseconds.  When {\tt DB\_SUBSCRIBE} is true, the value of this entry will
be sent from the manager to the client every {\tt rate} milliseconds.  If {\tt
periodic} is false, then when {\tt DB\_PUBLISH} is true, a "write" to this
entry will be propagated to the manager immediately, unless the last write was
less than {\tt rate} milliseconds ago.  Quick changes will be delayed for {\tt
rate} milliseconds.  When {\tt DB\_SUBSCRIBE} is true, changes made by the
manager will be propagated to the client immediately, unless the last change
was less than {\tt rate} milliseconds ago.  Note that {\tt rate} for
non-periodic access can be 0, which means changes are immediately propagated no
matter what, but this should be used with caution for high-priority entries
that will probably not change often (e-stops for example).

You cannot have both the {\tt DB\_PUBLISH} flag and the {\tt DB\_SUBSCRIBE}
flag true at the same time.  If this is the case, the entry request will be
rejected and NULL will be returned.

\begin{prototype}
RemoteDBEntry* create(bool periodic, const char* name, int max\_size, const char* description, unsigned int access\_flags, int rate, int timeout=500)
\end{prototype}

Subscribe to an entry on manager, and create it if it is not there already.
Returns 0 if successful, -1 if not.  Otherwise, this has the same semantics as
the {\tt access} method.

\begin{prototype}
RemoteDBEntry* accessInt(bool periodic, const char* name, unsigned int access\_flags, int rate, int timeout=500)
RemoteDBEntry* accessDouble(bool periodic, const char* name, unsigned int access\_flags, int rate, int timeout=500)
RemoteDBEntry* createInt(bool periodic, const char* name, unsigned int access\_flags, int rate, int timeout=500)
RemoteDBEntry* createDouble(bool periodic, const char* name, unsigned int access\_flags, int rate, int timeout=500)
\end{prototype}

These are convenience methods for creating and accessing remote database
entries which are integers and double precision floating point numbers.


\begin{prototype}
RemoteDBEntry* ACCESS(periodic, name, type, access\_flags, rate, timeout)
RemoteDBEntry* CREATE(periodic, name, type, access\_flags, rate, timeout)
\end{prototype}

These look like methods, but they are actually convenience macros for accessing
and creating entries using the same calling semantics as the {\tt create} and
{\tt access} methods, but whose size and descriptions are are extracted using
macro magic from the type name {\tt type}.

\subsubsection{Database client example}

% need to get focused example here
{\tt Needs to be done: Take a look at RHexLib/base/db\_test.cc}

\section{Logging Services}

logging services are designed to make it simple for an operator to collect
periodically sampled data from an operating robot and download the resulting
data to a client computer (typically part of the operator control unit or OCU).
The service is designed to be minimally invasive on the robot, but to enable the
effective collection of large quantities of data.  The service is divided into
two section: first is the \texttt{LogServer} class which operates on the robot
and is responsible data acquisition and buffering (see Section
\ref{sec:logserver}), second the \texttt{LogClient} class which abstracts the
underlying communication tasks needed to interact with the \texttt{LogServer}.
It is this latter class that we focus on in this section.

Note that while the \texttt{LogClient} and \texttt{LogServer} pair are designed
to support multiple client connections to a single server, there are limits in
the underlying protocol that prevent certain tasks from being performed
``simultaneously'' by multiple clients.  In particular starting a new logging
task can involve multiple messages sent between the client and server, and this
sequence can be ``interrupted'' by messages from other clients.  This said, if
multiple clients wish to interact with the robot the best procedure to follow is
to ensure that \texttt{startLog} calls are performed only when no other requests
are being sent to the server.

\subsection{LogClient interface}
The \texttt{LogClient} class provides the basic interface between a client
program (one that wishes to collect data from the robot) and the
\texttt{LogSever} class running on the robot.  This class includes various
methods for interacting with the \texttt{LogServer} class as well as methods for
creating and destroying logging tasks.  The typical sequence of operations is to
create an instance of the \texttt{LogClient} class, query the remote
\texttt{LogServer} class for a list of loggable variables, create a new logging
task, add the desired variable to the newly crated task, start the task, receive
and process/save/interpret the data from the logging task, stop the logging
task, destroy the logging task, and finally finish interaction with the
\texttt{LogServer} by destroying the \texttt{LogClient} instance.

Note: most methods accept an option argument \texttt{msec}, this argument can be
used to specify the desired timeout that the \texttt{LogClient} will wait for a
response from the associated \texttt{LogSerrver} before indicating failure.  The
default timeout is normally 5 seconds.

The specific interface to \texttt{LogClient} follows:
\begin{codesegment}
#include "LogClient.hh"
\end{codesegment}

\begin{classdef}
class LogClient {
public:
  LogClient(CommManager *com, RemoteManager *rmt,
            const com_id_t log_port=DEFAULT_LOG_STREAM);
  LogClient(CommManager *com,
            const char* host, const unsigned short port,
            const com_id_t log_port=DEFAULT_LOG_STREAM);

  ~LogClient();

  // BASIC API -- methods that send simple messages
  bool ping(const unsigned int msec=LOG_CLIENT_TIMEOUT);
  bool reset(const unsigned int msec=LOG_CLIENT_TIMEOUT);

  // VARIABLE MANAGEMENT -- methods for managing avaliable variables
  bool query(const unsigned int msec=LOG_CLIENT_TIMEOUT);
  void clearVars();
  unsigned int listVars();
  log_name_t *getNextVar();
  ListIterator<log_name_t> *varIterator();

  // LOG-JOB MANAGEMENT
  LogTask *newLog();
  bool destroyLog(LogTask *task);
};
\end{classdef}

The two constructors provide two alternate means to specify the computer to be
contacted, either as a pointer to a \texttt{RemoteManager} instance, or by
hostname and port number.  The final argument specifies the port number of the
stream where the \texttt{LogServer} is expecting to be contacted -- by default
the \texttt{LogServer} waits on stream port 76.  The basic API section includes
a \texttt{ping} method, which can be used to verify the health of the remote
\texttt{LogServer} -- note that the underlying protocol is unreliable and thus
while a successful ping (returning \texttt{true}) indicates that the
\texttt{LogServer} is running, a failed ping (returning \texttt{false}) should
not be used to conclude that the \texttt{LogServer} has failed\footnote{The
  \texttt{LogServer} replies to all messages by sending a status message to a
  mailbox since these could be dropped/lost it is general not safe to use any
  ``failed'' request to conclude that the \texttt{LogServer} has actually
  failed.  Rather, failed requests should be used only to indicate that there
  may be either network or software problems in the system.}.  The second method
in the group, \texttt{reset}, can be used to force the remote \texttt{LogServer}
to fully reset its internal state.  This has the effect of terminating all
active logging tasks, and breaking all pending connections/communications.

The variable management section provides methods for interrogating the available
variables from the \texttt{LogServer}.  The \texttt{LogClient} class maintains
an internal list of available variables.  This list is updated by calling
\texttt{query}.  When \texttt{query} is called, a list of available variables
is retrieved from the \texttt{LogServer} and the contents of this list are
merged with the existing internal list.  \texttt{clearVars} clears the contents
of the internal list.  \texttt{listVars} produces an alphabetically sorted list
of the available variables, and indexes an iterator to the beginning of this list,
\texttt{getNextVar} can then be used to extract the names of each
variable\footnote{See \texttt{LogMsg.h} for the definition of
  \texttt{log\_name\_t} which includes the variable name as well as type and size
  information}.  So a typical listing of the available variables can be generated
as follows.
\begin{codesegment}
  LogClient* logger = new LogClient(...);

  if (!logger->query()) {
    printf("Unable to query LogServer\n");
  } else {
    unsigned int num_var = logger->listVars();
    printf("%d variables avaliable\n", num_var);

    log_name_t *n;
    while((n=logger->getNextVar()))
      printf("<%s> %s[%d]\n",
             n->name, format_type(n->type), n->array);
  }
  delete logger;
\end{codesegment}
For convenience, access is also given to this sorted list via a
\texttt{ListIterator} -- note that the elements pointed to on this list are
internal to the \texttt{LogClient} class, and \textbf{must not} be changed by
the user.

Finally there are two methods for creating and destroying instances of the class
\texttt{LogTask}.  A \texttt{LogTask} instance represent a logging task,
specifying wich variables are to be collected, how frequently they should be
sampled, and for what period that sampling should continue.  Since a logging
task is inexorably intertwined with its associated \texttt{LogClient} instances
of \texttt{LogTask} are created and destroyed respectively with the
\texttt{newLog} and \texttt{destroyLog} methods of \texttt{LogClient}.

\subsection{LogTask interface}

The \texttt{LogTask} class provides an interface through which a client program
can specify and control the execution of a particular logging task.  A single
instance of \texttt{LogClient} will support and essentially arbitrary number of
\texttt{LogTask}s.  After creation of an instance of \texttt{Logtask} (as
described above) typically a program ``adds'' variable to be logged to the task,
then issues a ``start'' command, specifying the period, duration, and buffering
behavior for the task.  The specific class definition includes the following
methods: 

\begin{codesegment}
#include "LogClient.hh"
\end{codesegment}

\begin{classdef}
class LogTask {
public:
  bool addVar(const char *name);
  unsigned int startLog(const unsigned int period, const unsigned int length,
                        const unsigned int buf_size=1000,
                        const unsigned int msec=LOG_CLIENT_TIMEOUT);
  bool stopLog(const unsigned int msec=LOG_CLIENT_TIMEOUT);
  bool abortLog(const unsigned int msec=LOG_CLIENT_TIMEOUT);

  log_line_t *getData(const unsigned int msec=LOG_CLIENT_TIMEOUT);

  ListIterator<log_name_t> *varIterator();
  List<log_name_t> *varList();

  unsigned int totalData() const { return _data_size; };
  unsigned int bufferSize() const { return _buf_size; };
  unsigned int winSize() const { return _win_size; };
  bool isDone() const { return _data_done; };
\end{classdef}

Note that the constructor and destructor for this class are private, since an
instance of \texttt{LogTask} without an associated \texttt{LogClient} is
meaningless.  The method \texttt{addVar} is used to specify (by name) the
variables that should be logged for this task, note that every logging task
automatically timestamps all data collected with a 64-bit integer, representing
microseconds since the robot began operation.  \texttt{startLog} is used to
actually begin the logging process, here the logging period (in miliseconds), the
length or duration of the logging task (in number of samples to be collected),
the maximum buffer size (in units of approximately 1kB), and the timeout.
Specifying a length of 0 results in a task that will run until it is manually
stopped.  Once a job is started, \texttt{getData} can be used to retrieve data
from the task.  Each call to \texttt{getData} will return one line of data (the
value of each variable at one instant in time).  The format of this data is
defined in \texttt{LogMsg.hh} as
\begin{codesegment}
typedef struct {
  unsigned long long time;                        // timestamp
                                                  // -1 here indicates lost data
  unsigned char data[4];                          // start of the data
} log_line_t;
\end{codesegment}
The \texttt{time} field is the timestamp for this line of data, and the
\texttt{data} field marks the beginning of the actual data.  The variables
collected will overrun the \texttt{data} field and are packed tightly. in their
native format.  If either the end of the data stream has been reached or a
timeout occurs \texttt{getData} will return \texttt{NULL}, in either case the
status of the buffers should be checked (as described below) to determine that
cause.  In the event that the \texttt{LogServer} ran out of buffer space an
timestamp of -1 will be used to indicate where data was discarded.

Two methods for stopping a \texttt{LogTask} are included.  The first,
\texttt{stopLog}, causes an orderly shutdown of the logging task by sending
a message requesting that the \texttt{LogServer} cease collecting data, any data
already collected is buffered and returned to the client so that it may be
received via calls to \texttt{getData}.  The alternate method,
\texttt{abortLog}, causes an immediate stop of the logging task, all buffers are
discarded and the server is notified that the task should be discarded.  

Four functions are included to facilitate monitoring of a logging task.  the most
basic is \texttt{isDone}, this method returns true if all of the data associated
with the task has been retrieved from the \texttt{LogServer}.  \texttt{isDone}
returning true in conjunction with \texttt{getData} returning \texttt{NULL}
indicates that all data has been received and processed.  The methods
\texttt{totalData}, \texttt{bufferSize}, and \texttt{winSize} indicate
respectively the total amount of data received by the client process, the amount
of buffered data waiting to be processed by \texttt{getData}, and the amount of
mis-ordered data waiting to be sorted and buffered.  It is the last of these
three that deserves some addition explanation.  Since the underlying
communications protocol is unreliable, the \texttt{LogTask} class makes an
effort to ensure that the received data is presented to a user in the order it
was acquired, this requires extra ``sorting'' of the incoming data if message
packets are lost in transmission.  Basically a ``window buffer'' is used to
store packets that are out of order until the intervening packets arrive at
which point they can be safely appended to the end of the buffer.  The result
is a reliable ordered stream of logging data.

Finally two means of gaining access to the list of variables associated with the
logging task are provided.  The first return an iterator for the internal list,
while the second returns a pointer to the list itself.  In either case it is
critical that the contents of these lists not be modified in any way.




