usyncclient.hh

Go to the documentation of this file.

/*
 * Copyright (C) 2004, 2006-2011, Gostai S.A.S.
 *
 * This software is provided "as is" without warranty of any kind,
 * either expressed or implied, including but not limited to the
 * implied warranties of fitness for a particular purpose.
 *
 * See the LICENSE file for more information.
 */



#ifndef URBI_USYNCCLIENT_HH
# define URBI_USYNCCLIENT_HH

# include <libport/finally.hh>
# include <libport/fwd.hh>
# include <libport/lockable.hh>
# include <libport/semaphore.hh>
# include <libport/utime.hh>
# include <libport/pthread.h>

# include <urbi/uclient.hh>

namespace urbi
{
  enum UTransmitFormat
  {
    URBI_TRANSMIT_JPEG,
    URBI_TRANSMIT_YCbCr
  };


  class URBI_SDK_API USyncClient: public UClient
  {
  public:
    typedef boost::function1<void, USyncClient*> connect_callback_type;
    typedef UClient super_type;
    typedef super_type::error_type error_type;

    struct URBI_SDK_API options
      : public super_type::options
    {
      options();
      UCLIENT_OPTION(bool, startCallbackThread);
      UCLIENT_OPTION(connect_callback_type, connectCallback);
    };

    USyncClient(const std::string& host,
                unsigned port = URBI_PORT,
                size_t buflen = URBI_BUFLEN,
                const USyncClient::options& opt = options());

    ~USyncClient();

    struct send_options
    {
      send_options();

      send_options& timeout(libport::utime_t);
      send_options& tag(const char*, const char* = 0);

      libport::utime_t timeout_;
      const char* mtag_;
      const char* mmod_;
      static const send_options default_options;
    };

  protected:
    virtual error_type onClose ();

  protected:
    UMessage*
    syncGet_(const char* expression, va_list& arg,
             const send_options& options = send_options::default_options);

  public:
    ATTRIBUTE_PRINTF(2, 3)
    UMessage* syncGet(const char* expression, ...);

    UMessage* syncGet(const std::string& exp);

    ATTRIBUTE_PRINTF(3, 4)
    UMessage* syncGet(libport::utime_t useconds,
                      const char* expression, ...);
    ATTRIBUTE_PRINTF(2, 5)
    UMessage* syncGetTag(const char* expression,
                         const char* mtag, const char* mmod, ...);
    ATTRIBUTE_PRINTF(3, 6)
    UMessage* syncGetTag(libport::utime_t useconds,
                         const char* expression,
                         const char* mtag, const char* mmod, ...);

    int syncSend(const void* buffer, size_t length);

    int syncGetImage(const char* cameraDevice, void* buffer,
                     size_t& buffersize,
                     int format, int transmitFormat,
                     size_t& width, size_t& height,
                     libport::utime_t useconds = 0);

    int syncGetValue(const char* valName, UValue& val,
                     libport::utime_t useconds = 0);
    int syncGetValue(const char* tag, const char* valName, UValue& val,
                     libport::utime_t useconds = 0);

    int syncGetDevice(const char* device, ufloat &val,
                      libport::utime_t useconds = 0);

    int syncGetResult(const char* command, ufloat &val,
                      libport::utime_t useconds = 0);

    int syncGetNormalizedDevice(const char* device, ufloat &val,
                                libport::utime_t useconds = 0);

    int syncGetDevice(const char* device, const char* field, ufloat &val,
                      libport::utime_t useconds = 0);

    int syncGetSound(const char* device, int duration, USound &sound,
                     libport::utime_t useconds = 0);

    UMessage* waitForTag(const std::string& tag, libport::utime_t useconds = 0);

    void lockQueue();

    virtual void notifyCallbacks(const UMessage &msg);

    bool processEvents(libport::utime_t timeout = -1);

    void stopCallbackThread();
    void callbackThread();

    void setSynchronous(bool enable);
    using super_type::waitForKernelVersion;
    void waitForKernelVersion(bool hasProcessingThread);

    void setDefaultOptions(const USyncClient::send_options& opt);
    const USyncClient::send_options&
      getOptions(const USyncClient::send_options& opt =
                 USyncClient::send_options::default_options) const;

    static boost::shared_ptr<libport::Finally>
    listen(const std::string& host, const std::string& port,
           boost::system::error_code& erc,
           connect_callback_type connectCallback,
           size_t buflen = URBI_BUFLEN,
           bool startCallbackThread = true);

    virtual void onConnect();

    bool isCallbackThread() const;
  protected:
    int joinCallbackThread_();

    static libport::Socket* onAccept(connect_callback_type l, size_t buflen,
                                     bool startThread);
    // Incremented at each queue push, decremented on pop.
    libport::Semaphore sem_;
    // Semaphore to delay execution of callback thread until ctor finishes.
    libport::Semaphore callbackSem_;

    // The list of incoming messages waiting to be processed.
    std::list<UMessage*> queue;
    libport::Lockable queueLock_;

    UMessage* message_;
    libport::Semaphore syncLock_;
    std::string syncTag;

    send_options default_options_;

    bool stopCallbackThread_;
    pthread_t cbThread;
    // Used to block until the callback thread is realy stopped.
    libport::Semaphore stopCallbackSem_;
    connect_callback_type connectCallback_;
    // True if waitForTag is waiting from the polling thread.
    bool waitingFromPollThread_;
    // If set, bypass callback thread and send messages synchronously.
    bool synchronous_;
  };

} // namespace urbi

#endif // ! URBI_USYNCCLIENT_HH