streams.h

#ifndef STREAMS_H
 
#include "main-header.h"
#include "command-parser.h"
#include "command-handler.h"
#include "queue.h"
//#include "loader-api.h"
//#include "userhooks.h"
//#include "constants.h"
//#include "function-instr.h"
//#include "baseinfo.h"
 
#define STREAMS_H 1
typedef struct stream {
  byte                          fd;                     /* 0 -> NOT a valid stream yet */
  long                          closetn;                /* >0 received close request in this tn (but missing data) */
  byte                          pad;
  byte                          failed;                 /* if not 0 -> stream has failed */
  byte                          ackfreq;                /* after how many packets must we sent an ack? */
  byte                          ack_open;               /* set to 1 by sending an ACK, set to 0 by receiving data >ack */
  byte                          inactivity_ack_ctr;     /* set to 0 by good data, increased if no activity */
  byte                          outoforderctr;          /* incremented each time an out-of-order packet arrives and reset to 0 when packet is received in order or after sending an ack due to to too many out-order packets */
  int                           ackedtn;                /* most recent number we acked */
  int                           lasttn;                 /* most recent number we received */
  long                          peer;                   /* nodeid who is this stream connected to? */
  long                          lastack;                /* secs since last ack */
  long                          lastackms;              /* ms since last ack */
  long                          last_activity;          /* secs since last received */
  /* call consumer when we got this amount of bytes ready. if != 1, must set buffer as well */
  int                           packetsize;
  byte *                        bytebuf;
  /* buffer for packetisation (must not be not-null if >1 packetisation is requested) */
  byte                          onebytebuf;
  /* how many bytes are currently in buf? */
  int                           bytesinbuf;
  int                           bufsize;        /* buffer must be at least as big as packet size - bigger is better ;) */
  const struct consumerinfo *   consumer;       /* pointer to consumer of this stream */
  char                          target[18];
} _streamdef;
 
 
typedef int (*stream_consume_data)(int streamfd, byte *b, int size);
typedef int (*stream_consume_close)(int streamfd, int reason);
 
typedef struct consumerinfo {
  /* byte consumer if we have one */
  stream_consume_data   data_consumer;
  stream_consume_close  close_consumer;
} _consumerinfo;
 
 
void stream_init();
void streamdata(struct command *com);
void streamsetup(struct command *com);
void stream_loop();
void stream_disable();
void stream_enable();
void print_stream_info();
int stream_associate(int fd, const struct consumerinfo *c, byte *buf, int bufsize, int packetsize);
struct stream *alloc_stream();
void close_stream(int fd, int reason);
int count_open_streams();
/*
 * a stream consumer might encounter an error - can enter short message here
 */
void stream_seterror(int streamfd, const char *format, ...);
#endif