ngcbSTREAM.hpp

Go to the documentation of this file.

#ifndef ngcbSTREAM_H
#define ngcbSTREAM_H
 
 
#ifndef __cplusplus
#error This is a C++ include file and cannot be used from plain C
#endif
 
/* 
 * System Headers 
 */
 
/*
 * Local Headers 
 */
#include <ngc/core/ngcbTHREAD.hpp>

struct ngcb_fb_t {
  FILE *chan;

  int automatic;

  ngcb_fb_t() {chan = NULL; automatic = 0;}
};

class ngcbSTREAM: public ngcbTHREAD
{
public:
  explicit ngcbSTREAM(FILE *fd) {
    /*
     * Initialize data members
     */
    Init_(fd);
        
    /*
     * Create streaming process
     */
    stream_ = NewThread(this, (ngcbPROCESS_T)&ngcbSTREAM::Process_, NULL);
    if (stream_ == NULL)
      {
      sprintf(ErrMsg(), "unable to start streaming thread");
      return;
      }
    else if (!stream_->Initialized())
      {
      delete stream_;
      stream_ = NULL;
      sprintf(ErrMsg(), "unable to initialize streaming thread");
      return;
      }
    
    /*
     * Initialized
     */
    initialized_ = 1;
  }

  ngcbSTREAM(FILE *fd, ssize_t bufferSize) {
    /*
     * Initialize data members
     */
    Init_(fd);
 
    if (bufferSize <= 0)
      {
      if (bufferSize == 0)
    {
    /*
     * Disable streaming buffer
     */
    setbuf(fd, NULL);
    }
      }
    else
      {
      /*
       * Allocate buffer
       */
      buffer_ = (char *)malloc(bufferSize);
      if (buffer_ == NULL)
    {
    bufferSize_ = 0;
    sprintf(ErrMsg(), "unable to allocate streaming buffer");
    return;
    }
      bufferSize_ = bufferSize;
    
      /*
       * Create streaming process
       */
      stream_ = NewThread(this, (ngcbPROCESS_T)&ngcbSTREAM::Process_, NULL);
      if (stream_ == NULL)
    {
    free(buffer_);
    buffer_ = NULL;
    bufferSize_ = 0;
    sprintf(ErrMsg(), "unable to start streaming thread");
    return;
    }
      else if (!stream_->Initialized())
    {
    delete stream_;
    stream_ = NULL;
    free(buffer_);     
    buffer_ = NULL;
    bufferSize_ = 0;
    sprintf(ErrMsg(), "unable to initialize streaming thread");
    return;
    }
      }
 
    /*
     * Initialized
     */
    initialized_ = 1;
  }

  virtual ~ngcbSTREAM() {
    /*
     * Wait for transfer
     */
    if (transfer_ && (semAck_ != NULL)) semAck_->Wait();
 
    /*
     * Enforce streaming process termination
     */
    if (semSig_ != NULL) semSig_->Post();
 
    /*
     * Free resources
     */
    if (stream_ != NULL) delete stream_; 
    if (semSig_ != NULL) delete semSig_; 
    if (semAck_ != NULL) delete semAck_; 
    if (buffer_ != NULL) free(buffer_); 
  }

  size_t Write(const char *buffer, size_t size) {
    size_t tx;
    size_t b2w = size;
    size_t offset = 0;
 
    if (bufferSize_ == 0)
      {
      /*
       * System write
       */
      if (fwrite(buffer, 1, size, fd_) != size)
    {
    strcpy(ErrMsg(), strerror(errno));
    return (0);
    }
      
      return (size);
      }
    
    if (stream_ == NULL)
      {
      strcpy(ErrMsg(), "invalid stream");
      return (0);
      }
 
    while (b2w > 0)
      {
      if (b2w <= bufferSize_) tx = b2w;
      else tx = bufferSize_;
      
      /*
       * Wait until stream is free
       */
      semAck_->Wait();
      transfer_ = 0;
 
      /*
       * Check if an error has occurred
       */
      if (error_)
    {
    /*
     * Delete the stream
     */
    delete stream_; stream_ = NULL;
    return (0);
    }
 
      /*
       * Copy next chunk to streaming buffer
       */
      memcpy(buffer_, buffer + offset, tx);
 
      /*
       * Initiate transfer
       */
      size_ = tx;
      source_ = (const char *)buffer_;
      transfer_ = 1;
      semSig_->Post();
 
      b2w -= tx;
      offset += tx;
      }
    
    return (size);
  }
  
  int Send(const char *buffer, size_t size) {
 
    if (stream_ == NULL)
      {
      strcpy(ErrMsg(), "invalid stream");
      return (-1);
      }
 
    /*
     * Wait until stream is free
     */
    semAck_->Wait();
    transfer_ = 0;
 
    /*
     * Check if an error has occurred
     */
    if (error_)
      {
      /*
       * Delete the stream
       */
      delete stream_; stream_ = NULL;
      return (-1);
      }
 
    /*
     * Pass the buffer address to streaming process
     */
    source_ = buffer;
    
    /*
     * Initiate transfer
     */
    size_ = size;
    transfer_ = 1;
    semSig_->Post();
    
    return (0);
  }

  char *ErrMsg() {return (erms_);}

  int Initialized() {return (initialized_);}
 
protected:
 
private:
  char erms_[256];

  int error_;

  int initialized_;

  int transfer_;

  ngcbTHREAD *stream_;

  ngcbSEM *semSig_;

  ngcbSEM *semAck_;

  char *buffer_;

  const char *source_;

  size_t bufferSize_;  

  size_t size_;

  FILE *fd_;

  void Init_(FILE *fd) {
    initialized_ = 0;
    error_ = 0;
    transfer_ = 0;
    erms_[0] = '\0';
    fd_ = fd;
    stream_ = NULL;
    buffer_ = NULL;
    source_ = NULL;
    bufferSize_ = 0;
    size_ = 0;
    semSig_ = new ngcbSEM(0);
    semAck_ = new ngcbSEM(0);
  }

  void Process_(ngcbTHREAD *thr, void *arg) {
 
    /*
     * Process initialized
     */
    semAck_->Post();
 
    while (!error_)
      {
      /*
       * Wait for next transfer signal
       */
      semSig_->Wait();
 
      if (size_ == 0)
    {
    /*
     * Terminate
     */
    semAck_->Post();
    break;
    }
 
      if (source_ != NULL)
    {
    /*
     * Write buffer to stream
     */
    if (fwrite(source_, 1, size_, fd_) != size_)
      {
      strcpy(ErrMsg(), strerror(errno));
      error_ = 1;
      }
    }
      else
    {
    strcpy(ErrMsg(), "invalid buffer source (NULL)");
    error_ = 1;
    }
 
      /*
       * Acknowledge transfer
       */
      semAck_->Post();
      }
 
    ThrExit();
  }
 
};
 
#endif