CScommBase.hh

/*
This file is part of the BIAS library (Basic ImageAlgorithmS).

Copyright (C) 2003-2009    (see file CONTACT for details)
Multimediale Systeme der Informationsverarbeitung
Institut fuer Informatik
Christian-Albrechts-Universitaet Kiel


BIAS is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 2.1 of the License, or
(at your option) any later version.

BIAS is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with BIAS; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#ifndef __BIASCScommBase__
#define __BIASCScommBase__

#include <bias_config.h>
#include <Base/Common/W32Compat.hh>
#include <Base/Common/BIASpragmaStart.hh>

#ifndef BIAS_HAVE_PTHREADS
#  error BIAS_HAVE_PTHREADS not defined. But BIASNetworkComm needs pthreads (on Unix and Win32). Please enable USE_PTHREADS in CMake configure step.
#endif

#include <pthread.h>
#include <sys/types.h>
#include <vector>
#include <map>
#include <Base/Debug/Debug.hh>
#include <NetworkComm/CScommClasses.hh>

namespace BIAS {

#define D_CS_MAXCLIENTS 100
#define D_CS_DEFAULT_PORT 4123

#define D_CS_LISTENER         1<<0
#define D_CS_RECEIVER         1<<1
#define D_CS_CONNECT          1<<2
#define D_CS_ANALYZE          1<<3
#define D_CS_SEND             1<<4
#define D_CS_GETDATA          1<<5

#ifdef WIN32
# define mygetpid() (pthread_self()).p
#else
# define mygetpid() getpid()
#endif //WIN32

  /** 
        @class CScommBase
        @brief this class CScomm (ClientServer communciation)
        handles data comunication via TCP/IP for one server and
        multiple clients.
        @ingroup network
        It spawns threads itself for all communication.
        Messages can be registered by functions or within a file(not yet).
        See CScommClient or CScommServer for more details.
    
        @author Daniel Grest, Jan 2004
  */
  class BIASNetworkComm_EXPORT CScommBase
    : public BIAS::Debug
  {
  public:
    virtual ~CScommBase();

    /** Set threaded to false for deactivating the receive-thread. Data will be
    read from TCP-buffer only when GetData methods are called. Threaded
    receive is activated by default!
    Beware: if there's no msg in tcp-buffer, GetData will BLOCK until there
    is any new msg (this need not be the one GetData was called for,
    for example binary instead of int)
    if receive-buffer is completly filled, send blocks until msg
    can be delivered!
    */
    CScommBase(bool threaded = true);

    /// copy ctor required due to ostream member with explicit ctor JW
    CScommBase(const CScommBase & arg);

    /// assignment op. required due to ostream member with explicit ctor JW
    CScommBase & operator=(const CScommBase & arg);

    /** If not threaded you can set this true to keep all received data.
    Else a call of GetData() will deliver the newest msg only.
    Default is off.
    If threaded this is useless... (always off)
    */
    void SetKeepAll(bool on) { keepAll_=on; }

    /** only if not threaded! <br>
    Blocks until any new msg arrives,
    or returns instantly if a new message
    is already in a buffer slot. The msg name is returned in msgName.

    The return value indicates how many additional new msgs are in the
    buffer slots, e.g. if 0 the msgName is the only new message.
    @return <0 if an error occured, >=0 the amount of new msgs in other
    slots.
    */
    int WaitForNewMessage(std::string &msgName);


    /** returns instantly:<br>
    0 if new data has arrived and the data in xxxData <br>
    1 if no new data has arrived, the data in xxxData stays unchanged<br>
    -1 in case of errors
    */
    int GetData(const std::string &msgName, std::vector<float> &floatData);
    int GetData(const std::string &msgName, std::vector<int> &intData);
    int GetData(const std::string &msgName,
      std::vector<std::string> &stringData);
    /** returns the binarydata in the vector, its wise to reserve enough
    bytes, if the size is known.
    */
    int GetData(const std::string &msgName,
      std::vector<char> &binaryData);

    /** sends msg msgName to all connected commPartners.
    Msg doesn't have to be registered on the sender side,
    but must be registered on the receiver side!
    */
    int SendMsg(const std::string &msgName, std::vector<float> &floatData);
    int SendMsg(const std::string &msgName, std::vector<int> &intData);
    int SendMsg(const std::string &msgName,
      std::vector<std::string> &stringData);

    /** maximum size for binary data is 15MB!
    */
    int SendMsg(const std::string &msgName, char* binaryData, unsigned int size);

    /** like above, but msg is only send to one specific commPartner given
    as string, which must equal the name given by the client to itself<br>
    @return -2 if no such commPartner, 0 in case of no error
    */
    int SendMsg(const std::string commPartnerName,
      const std::string &msgName, char* binaryData, unsigned int size);

    /** like above, but msg is only send to one specific commPartner given
    as the index number of all connected partners
    @return -2 if no such commPartner, 0 in case of no error
    */
    int SendMsg(int commPartnerNr,
      const std::string &msgName, char* binaryData, unsigned int size);


    /** only registered msgs are accepted by commPartners
    Register a msg with this functions.
    Amount is the number of dataTypes for this msg,
    for binarydata this is not necessary <br>
    The msg will be bound to the first commPartner sending such a msg.
    And will be accepted only from him later.<br>
    All msgNames have to be unique!

    return zero on success
    <0 on errors
    */
    int RegisterMsg(std::string msgName,  EdataType dataType, int amount=1);

    /** same as above, but the msg is only accepted from a specific commPartner
    */
    int RegisterMsg(std::string commPartnerName,
      std::string msgName,  EdataType dataType, int amount=1);

    /** returns the number of connected commPartners
    */
    inline int GetConnections() { return  commPartnersConnected_; }

    /** returns the name of the commPartner with nr commPartnerNr or
    empty string in case of a number, that is to high.
    */
    std::string GetCommPartnerName(int commPartnerNr);

    /** set the time in ms until a commPartner is assumed to be disconnected.<br>
    If time is zero there is no timeOut. default is no Time out
    */
    inline void SetConnectionTimeOut(unsigned int time) {
      if (time==0) timeOutFlag_=false; timeOut_=time;}

    /** if set on, the server will take automatically the next free TCP port
    for connections. The client tries 10 consecutive different ports
    to conncect the server. Default is off. Useful if you don't
    care about port numbers and just want a connection somehow.
    */
    inline void SetAutomaticPortChange(bool on) { portChange_=on; }

    // tells wether receive-threads are active or not
    inline bool GetThreadedReceive() { return threaded_; }

    /** gives a lot of information regarding the connections, like the exact
    data received and send. Implies verbose.
    */
    void SetLog(bool on);

    /** set the output of log msgs to another stream, e.g. to a file.
    Default ist cout.
    */
    void SetLogOut(std::ostream &outputStream);

    /** gives some information about establising conn, and disconnecting etc.
    */
    void SetVerbose(bool on);

    /** set the output of verbose msgs to another stream, e.g. to a file.
    Default ist cout.
    */
    void SetVerboseOut(std::ostream &outputStream);

  protected:

    // exports for dll (cmenk)
    //BIAS_EXPORT_STL_VECTOR( BIASNetworkComm_EXPORT, CScommData )
    //BIAS_EXPORT_STL_VECTOR( BIASNetworkComm_EXPORT, CScommMsg )
    //BIAS_EXPORT_STL_MAP( BIASNetworkComm_EXPORT, std::string, int )

    void logOutPut_(const char *data, unsigned int size);

    // helper func for the SendMsg() funcs
    int SendMsg_(const std::string &data);

    // helper for GetData(), finds the index of the msg in registeredMsgs
    // returns the index, by use of a the std::map indexRegMsgs_
    int GetData_(const std::string &msgName);

    /** return values:
    0 no error all fine

    warnings:
    1 not enough data to analyze (<3 bytes)
    2 no data after msg name
    3 more data read than registered for a msg
    4 less data read than registered
    A warning is returned if one of the above cases was true for any one
    msg in the buffer.

    errors:
    -1 only trash received
    */
    int AnalyzeData_(int clientNr);

    // checks if the msg with msgName ist registered for this comm and
    // returns the dataTypeToread and the dataSizeToRead if the msg is registered
    // return value is the index of the reg msg in registeredMsgs_ ,
    // or <0 for errors and unregistered msgs
    int CheckRegisteredMsgs_(std::string &msgName, const CScommData &cPartner,
      EdataType &dataTypeToRead,
      unsigned int &dataSizeToRead);

    void receiveThread_(int clientNr);

    //Replaces receiveThread_(..) if threaded-receive is deactivated.
    //Reads up to 200000 bytes from receive-buffer for all comm-partners
    void syncReceive_();

    // this is specific for server, dataSize is changed to the rest of bytes to
    // process, while offset is the number of processed bytes
    int AnalyzeFirstMsg_(int commNr);

    // starts a receive thread for a new connection, called by listener(server)
    // or by ConnectServer (client)
    int AcceptConn_(int fd);

    std::vector<CScommData> commPartners_;
    int connectPort_;
    int commPartnersConnected_;

    // if true, the TCP input buffer is read completely each time
    bool onlyNewMsgs_;
    bool isServer_;

    // added by aptersen
    bool portChange_;
    bool threaded_;
    bool keepAll_;

    bool log_;
    std::ostream logOut_;
    bool verbose_;
    std::ostream verboseOut_;

    std::string name_;
    unsigned int timeOut_;//after this time(ms) without data ,a comm is disc.
    bool timeOutFlag_; // if false, there is no timeOut
    bool terminateReceiver_;
    pthread_mutex_t registerLock_; // needed for acces on registeredMsgs_

    std::vector<CScommMsg> msgBuffer_;
    std::vector<CScommMsg> registeredMsgs_;

  private: // avoid DLL export
    /// used by GetData
    /// stores the index to registeredMsgs_
    std::map<std::string, int> indexRegMsgs_;

    //  bool binaryMsgActive_; // for analyzeData

  protected:
    friend void* receivethread(void *data);
    //  friend void* listenthread(void *data);
  };

  /// these are wrapper to comply to the pthread_create() interface
  void *receivethread(void *data);
  //extern void* listenthread(void *data); // defined in CScommServer.cc


}
#include <Base/Common/BIASpragmaEnd.hh>
#endif