PeerCast.root/PeerCast/core/common/cstream.h

   1 // ------------------------------------------------
   2 // File : cstream.h
   3 // Date: 12-mar-2004
   4 // Author: giles
   5 //
   6 // (c) 2002-4 peercast.org
   7 // ------------------------------------------------
   8 // This program is free software; you can redistribute it and/or modify
   9 // it under the terms of the GNU General Public License as published by
  10 // the Free Software Foundation; either version 2 of the License, or
  11 // (at your option) any later version.
  12
  13 // This program is distributed in the hope that it will be useful,
  14 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 // GNU General Public License for more details.
  17 // ------------------------------------------------
  18
  19 #ifndef _CSTREAM_H
  20 #define _CSTREAM_H
  21
  22 // ----------------------------------
  23
  24 class Channel;
  25 class ChanPacket;
  26 class ChanPacketv;
  27 class Stream;
  28
  29
  30 // ----------------------------------
  31 class ChanPacket
  32 {
  33 public:
  34         enum
  35         {
  36                 MAX_DATALEN = 16384
  37         };
  38
  39         enum TYPE
  40         {
  41                 T_UNKNOWN       = 0,
  42                 T_HEAD          = 1,
  43                 T_DATA          = 2,
  44                 T_META          = 4,
  45                 T_PCP           = 16,
  46                 T_ALL           = 0xff
  47         };
  48
  49         ChanPacket()
  50         {
  51                 init();
  52         }
  53
  54         void    init()
  55         {
  56                 type = T_UNKNOWN;
  57                 len = 0;
  58                 pos = 0;
  59                 sync = 0;
  60                 skip = false;
  61                 priority = 0;
  62         }
  63         void    init(ChanPacketv &p);
  64         void    init(TYPE t, const void *, unsigned int , unsigned int );
  65
  66         void    writeRaw(Stream &);
  67         void    writePeercast(Stream &);
  68         void    readPeercast(Stream &);
  69
  70
  71         unsigned int sync;
  72         unsigned int pos;
  73         TYPE type;
  74         unsigned int len;
  75         char data[MAX_DATALEN];
  76         bool skip;
  77
  78         int priority;
  79 };
  80 // ----------------------------------
  81 class ChanPacketv
  82 {
  83 public:
  84         enum {BSIZE = 0x100};
  85         ChanPacketv()
  86         {
  87                 init();
  88         }
  89         ~ChanPacketv()
  90         {
  91                 free();
  92         }
  93
  94         void free()
  95         {
  96                 if (data) {
  97                         delete [] data;
  98                         data = NULL;
  99                         datasize = 0;
 100                 }
 101         }
 102         void reset()
 103         {
 104                 free();
 105                 init();
 106         }
 107         void    init()
 108         {
 109                 type = ChanPacket::T_UNKNOWN;
 110                 len = 0;
 111                 pos = 0;
 112                 sync = 0;
 113                 skip = false;
 114                 data = NULL;
 115                 datasize = 0;
 116                 priority = 0;
 117         }
 118         void init(ChanPacket &p)
 119         {
 120                 if (data && (datasize < p.len || datasize > p.len + BSIZE * 4)) {
 121                         free();
 122                         data = NULL;
 123                         datasize = 0;
 124                 }
 125                 type = p.type;
 126                 len = p.len;
 127                 pos = p.pos;
 128                 sync = p.sync;
 129                 skip = p.skip;
 130                 priority = p.priority;
 131                 if (!data) {
 132                         datasize = (len & ~(BSIZE - 1)) + BSIZE;
 133                         data = new char[datasize];
 134                 }
 135                 memcpy(data, p.data, len);
 136         }
 137         void init(ChanPacketv &p)
 138         {
 139                 ChanPacket tp;
 140                 tp.init(p);
 141                 init(tp);
 142         }
 143
 144         void    writeRaw(Stream &);
 145         void    writePeercast(Stream &);
 146         void    readPeercast(Stream &);
 147
 148         unsigned int sync;
 149         unsigned int pos;
 150         ChanPacket::TYPE type;
 151         unsigned int len;
 152         char *data;
 153         unsigned int datasize;
 154         bool skip;
 155
 156         int priority;
 157 };
 158 // ----------------------------------
 159 class ChanPacketBuffer
 160 {
 161 public:
 162         enum {
 163                 MAX_PACKETS = 64,
 164                 NUM_SAFEPACKETS = 60
 165         };
 166
 167         void    init()
 168         {
 169                 lock.on();
 170                 lastPos = firstPos = safePos = 0;
 171                 readPos = writePos = 0;
 172                 accept = 0;
 173                 lastWriteTime = 0;
 174                 for (int i = 0; i < MAX_PACKETS; i++) packets[i].reset();
 175                 lock.off();
 176
 177                 lastSkipTime = 0;
 178         }
 179
 180         int copyFrom(ChanPacketBuffer &,unsigned in);
 181
 182         bool    writePacket(ChanPacket &,bool = false);
 183         void    readPacket(ChanPacket &);
 184         void readPacketPri(ChanPacket &);
 185
 186         bool    willSkip();
 187
 188         int             numPending() {return writePos-readPos;}
 189
 190         unsigned int    getLatestPos();
 191         unsigned int    getOldestPos();
 192         unsigned int    findOldestPos(unsigned int);
 193         bool    findPacket(unsigned int,ChanPacket &);
 194         unsigned int    getStreamPos(unsigned int);
 195         unsigned int    getStreamPosEnd(unsigned int);
 196         unsigned int    getLastSync();
 197
 198         //ChanPacket    packets[MAX_PACKETS];
 199         ChanPacketv     packets[MAX_PACKETS];
 200         volatile unsigned int lastPos,firstPos,safePos;
 201         volatile unsigned int readPos,writePos;
 202         unsigned int accept;
 203         unsigned int lastWriteTime;
 204         WLock lock;
 205
 206         unsigned int lastSkipTime;
 207 };
 208
 209 // ----------------------------------
 210 class ChannelStream
 211 {
 212 public:
 213         ChannelStream()
 214         :numListeners(0)
 215         ,numRelays(0)
 216         ,isPlaying(false)
 217         ,fwState(0)
 218         ,lastUpdate(0)
 219         ,lastCheckTime(0)
 220         ,parent(NULL)
 221         {}
 222         virtual ~ChannelStream() {}
 223
 224         void updateStatus(Channel *);
 225         bool getStatus(Channel *,ChanPacket &);
 226
 227         virtual void kill() {}
 228         virtual bool sendPacket(ChanPacket &,GnuID &) {return false;}
 229         virtual void flush(Stream &) {}
 230         virtual unsigned int flushUb(Stream &, unsigned int) { return 0; }
 231         virtual void readHeader(Stream &,Channel *)=0;
 232         virtual int  readPacket(Stream &,Channel *)=0;
 233         virtual void readEnd(Stream &,Channel *)=0;
 234
 235         void    readRaw(Stream &,Channel *);
 236
 237         int             numRelays;
 238         int             numListeners;
 239         bool    isPlaying;
 240         int     fwState;
 241         unsigned int lastUpdate;
 242         unsigned int lastCheckTime;
 243
 244         Channel *parent;
 245 };
 246
 247 #endif
 248