extern "C" {

#include <my_base.h>

#define BSTREAM_USE_MALLOC
#include <stream_v1.h>
#include <stream_v1_services.h>

int bstream_write_blob(backup_stream *s, bstream_blob buf);
int bstream_read_blob(backup_stream *s, bstream_blob buf);
int bstream_end_chunk(backup_stream *s);

}

typedef bstream_byte byte;
typedef bstream_blob blob;

class Stream: public st_backup_stream
{
 public:

  Stream(const char *path);

  int open_wr(size_t block_size =0);
  int open_rd();
  int close();

 private:

  const char *m_path;
  int m_fd;

  static int write(void*, blob*, blob);
  static int read(void*, blob*, blob);
  static int forward(void*, unsigned long*);
};

#define CHUNK_SIZE 128
#define BLOCK_SIZE 10

int main()
{
  byte  buf[CHUNK_SIZE];
  blob  data;
  Stream str("bug45093.bst");

  memset(buf,0x5C, CHUNK_SIZE);
  buf[0]= 0xC5;

  if (str.open_wr(BLOCK_SIZE))
  {
    perror("Can not open stream for writing!");
    exit(2);
  };
  
  data.begin= buf;
  data.end=   buf + CHUNK_SIZE;

  bstream_write_blob(&str, data);
  bstream_end_chunk(&str);

  data.begin= buf;

  bstream_write_blob(&str, data);
  
  str.close();

  if (str.open_rd())
  {
    perror("Can not open stream for reading!");
    exit(2);
  };

  data.begin= buf;
  data.end=   buf + CHUNK_SIZE/4 + 1;

  bstream_read_blob(&str, data);
  bstream_next_chunk(&str);

  data.begin= buf;

  bstream_read_blob(&str, data);

  if (buf[0] != 0xC5 || buf[1] != 0x5C)
  {
    printf("Wrong data! (0x%02X,0x%02X)\n", buf[0],buf[1]);
    exit(1);
  }
  else
  {
    printf("Test OK\n");
    exit(0);
  }
}

/*====================================================================
 * 
 * Implementation of Stream class
 *  
 *====================================================================*/
 
Stream::Stream(const char *path): m_path(path)
{
  memset(&stream, 0, sizeof(stream));
}

int Stream::open_wr(size_t block_size)
{
  // open file for writing
  m_fd= open(m_path, O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR);
  if (m_fd < 0)
    return BSTREAM_ERROR;

  stream.write= write;
  stream.forward= forward;
  return bstream_open_wr(this, block_size, 0);
}

int Stream::open_rd()
{
  // open file for reading
  m_fd= open(m_path, O_RDONLY);
  if (m_fd < 0)
    return BSTREAM_ERROR;

  stream.read=  read;
  stream.forward= forward;
  return bstream_open_rd(this, 0);
}

int Stream::close()
{
  (void)bstream_close(this);
  // close the stream
  if (m_fd >= 0)
    ::close(m_fd);

  return BSTREAM_OK;
}

int Stream::write(void *stream, blob *buf, blob)
{
  int fd;
  int res;

  DBUG_ASSERT(stream);
  DBUG_ASSERT(buf);

  Stream *s= (Stream*)stream;

  fd= s->m_fd;

  DBUG_ASSERT(fd >= 0);

  if (!buf->begin || buf->begin == buf->end)
    return BSTREAM_OK;

  DBUG_ASSERT(buf->end);

  size_t howmuch = buf->end - buf->begin;
  
  while (howmuch)
  {
    ssize_t written= ::write(fd, buf->begin, howmuch);

    if (written <= 0)
      return BSTREAM_ERROR;

    buf->begin += written;
    howmuch -= written;
  }

  return BSTREAM_OK;
}

int Stream::read(void *stream, blob *buf, blob)
{
  int fd;
  size_t howmuch;

  DBUG_ASSERT(stream);
  DBUG_ASSERT(buf);

  Stream *s= (Stream*)stream;

  fd= s->m_fd;

  DBUG_ASSERT(fd >= 0);

  if (!buf->begin || buf->begin == buf->end)
    return BSTREAM_OK;

  DBUG_ASSERT(buf->end);

  howmuch= buf->end - buf->begin;
  
  while (howmuch)
  {
    ssize_t bytes_read= ::read(fd, buf->begin, howmuch);

    if (bytes_read == 0)
      return BSTREAM_EOS;

    if (bytes_read < 0)
      return BSTREAM_ERROR;

    buf->begin += bytes_read;
    howmuch -= bytes_read;
  }

  return BSTREAM_OK;
}

int Stream::forward(void *stream, unsigned long int *offset)
{
  int fd;
  size_t howmuch;

  DBUG_ASSERT(stream);
  DBUG_ASSERT(offset);

  Stream *s= (Stream*)stream;

  fd= s->m_fd;

  DBUG_ASSERT(fd >= 0);

  if (*offset == 0)
    return BSTREAM_OK;

  howmuch= *offset;

  lseek(fd, howmuch, SEEK_CUR);
  *offset= 0;

  return BSTREAM_OK;
}

/*====================================================================
 * 
 * Trivial implementatios of stream library callback functions
 *  
 *====================================================================*/

int bcat_reset(struct st_bstream_image_header *catalogue)
{ return BSTREAM_OK; }

int bcat_close(struct st_bstream_image_header *catalogue)
{ return BSTREAM_OK; }

int bcat_add_item(struct st_bstream_image_header *catalogue,
                  struct st_bstream_item_info *item)
{ return BSTREAM_OK; }

void* bcat_iterator_get(struct st_bstream_image_header *catalogue,
                        unsigned int type)
{ return NULL; }

struct st_bstream_item_info*
bcat_iterator_next(struct st_bstream_image_header *catalogue, void *iter)
{ return NULL; }

void  bcat_iterator_free(struct st_bstream_image_header *catalogue, void *iter)
{}

void* bcat_db_iterator_get(struct st_bstream_image_header *catalogue,
                           struct st_bstream_db_info *db)
{ return NULL; }

/** Return next item from database items iterator */
struct st_bstream_dbitem_info*
bcat_db_iterator_next(struct st_bstream_image_header *catalogue,
                      struct st_bstream_db_info *db,
                      void *iter)
{ return NULL; }

/** Free database items iterator resources */
void  bcat_db_iterator_free(struct st_bstream_image_header *catalogue,
                            struct st_bstream_db_info *db,
                            void *iter)
{}

int bcat_get_item_create_query(struct st_bstream_image_header *catalogue,
                               struct st_bstream_item_info *item,
                               bstream_blob *stmt)
{ return BSTREAM_OK; }

int bcat_get_item_create_data(struct st_bstream_image_header *catalogue,
                              struct st_bstream_item_info *item,
                              bstream_blob *data)
{ return BSTREAM_OK; }
                              
int bcat_create_item(struct st_bstream_image_header *catalogue,
                     struct st_bstream_item_info *item,
                     bstream_blob create_stmt,
                     bstream_blob other_meta_data)
{ return BSTREAM_OK; }