wvdotnet/wvstream.cs

   1 /*
   2  * Versaplex:
   3  *   Copyright (C)2007-2008 Versabanq Innovations Inc. and contributors.
   4  *       See the included file named LICENSE for license information.
   5  */
   6 using System;
   7 using System.Net;
   8 using Wv.Extensions;
   9
  10 namespace Wv
  11 {
  12     public interface IWvStream: IDisposable
  13     {
  14         bool ok { get; }
  15         Exception err { get; }
  16         EndPoint localaddr { get; }
  17         EndPoint remoteaddr { get; }
  18
  19         int read(WvBytes b);
  20         int write(WvBytes b);
  21         bool flush(int msec_timeout);
  22
  23         event Action onreadable;
  24         event Action onwritable;
  25         event Action onclose;
  26
  27         void close();
  28         void noread();
  29         void nowrite();
  30     }
  31
  32     public class WvStream : IWvStream
  33     {
  34         public static IWvEventer ev = new WvEventer();
  35
  36         public static void runonce()
  37         {
  38             ev.runonce();
  39         }
  40
  41         public static void runonce(int msec_timeout)
  42         {
  43             ev.runonce(msec_timeout);
  44         }
  45
  46         public WvStream()
  47         {
  48         }
  49
  50         public void Dispose()
  51         {
  52             close();
  53         }
  54
  55         bool is_readable = false, is_writable = false;
  56         event Action _onreadable, _onwritable, _onclose;
  57
  58         protected bool can_onreadable { get { return _onreadable != null; } }
  59         protected bool can_onwritable { get { return _onwritable != null; } }
  60
  61         protected void do_readable()
  62         {
  63             if (can_onreadable)
  64             {
  65                 is_readable = false;
  66                 _onreadable();
  67             }
  68         }
  69
  70         protected void do_writable()
  71         {
  72             if (can_onwritable)
  73             {
  74                 is_writable = false;
  75                 _onwritable();
  76             }
  77         }
  78
  79         protected void do_close()
  80         {
  81             if (_onclose != null)
  82                 _onclose();
  83         }
  84
  85         object pr_obj = new object();
  86         protected void post_readable()
  87         {
  88             is_readable = true;
  89             ev.addpending(pr_obj, do_readable);
  90         }
  91
  92         object pw_obj = new object();
  93         protected void post_writable()
  94         {
  95             is_writable = true;
  96             ev.addpending(pw_obj, do_writable);
  97         }
  98
  99         public virtual event Action onreadable {
 100             add    { _onreadable += value; if (is_readable) post_readable(); }
 101             remove { _onreadable -= value; }
 102         }
 103         public virtual event Action onwritable {
 104             add    { _onwritable += value; if (is_writable) post_writable(); }
 105             remove { _onwritable -= value; }
 106         }
 107         public virtual event Action onclose {
 108             add    { _onclose += value; }
 109             remove { _onclose -= value; }
 110         }
 111
 112         bool isopen = true;
 113         public virtual bool ok { get { return isopen && err == null; } }
 114
 115         Exception _err;
 116         public virtual Exception err {
 117             get {
 118                 return _err;
 119             }
 120             set {
 121                 if (_err == null) // remember the *first* error
 122                 {
 123                     _err = value;
 124                     close();
 125                 }
 126             }
 127         }
 128
 129         public virtual void close()
 130         {
 131             ev.delpending(pr_obj);
 132             ev.delpending(pw_obj);
 133             canread = false;
 134             flush(-1);
 135             canwrite = false;
 136             if (isopen)
 137             {
 138                 isopen = false;
 139                 if (_onclose != null) _onclose();
 140             }
 141             GC.SuppressFinalize(this);
 142         }
 143
 144         public virtual EndPoint localaddr {
 145             get { return null; }
 146         }
 147
 148         public virtual EndPoint remoteaddr {
 149             get { return null; }
 150         }
 151
 152         public virtual int read(WvBytes b)
 153         {
 154             return 0;
 155         }
 156
 157         // for convenience.  Note: always returns non-null, but the returned
 158         // array size might be zero.
 159         public WvBytes read(int len)
 160         {
 161             WvBytes bytes = new byte[len];
 162             int got = read(bytes);
 163             return bytes.sub(0, got);
 164         }
 165
 166         public void read(WvBuf b, int max)
 167         {
 168             int got = read(b.alloc(max));
 169             b.unalloc(max-got);
 170         }
 171
 172         public virtual int write(WvBytes b)
 173         {
 174             return b.len; // lie: we "wrote" all the bytes to nowhere
 175         }
 176
 177         /**
 178          * Wait up to msec_timeout milliseconds for the stream to become
 179          * readable or writable, respectively.
 180          *
 181          * Our default implementation always returns immediately, consistent
 182          * with the Unix select() behaviour of returning immediately on
 183          * non-select()able file handles.
 184          *
 185          * Returns true if the stream is readable or writable before the
 186          * timeout, false otherwise.
 187          *
 188          * Waiting synchronously is usually a bad idea in WvStreams
 189          * programs.  Use onreadable/onwritable/onclose instead whenever
 190          * you can.
 191          */
 192         public virtual bool wait(int msec_timeout,
 193                                  bool readable, bool writable)
 194         {
 195             if (readable && is_readable)
 196                 return true;
 197             if (writable && is_writable)
 198                 return true;
 199             if (!ok || (readable && !canread) || (writable && !canwrite))
 200                 return false;
 201             return true;
 202         }
 203
 204         // Don't make these anything but private!  They're tempting, but
 205         // they're not *really* what you want to know.  Use these instead:
 206         //    ok -> whether or not your stream is useful at all
 207         //    wait(0, true, false) -> whether your stream has bytes *right now*.
 208         // If canread goes false, wait(-1, true, false) will return false.
 209         bool canread = true, canwrite = true;
 210
 211         public virtual void noread()
 212         {
 213             canread = false;
 214             maybe_autoclose();
 215         }
 216
 217         public virtual void nowrite()
 218         {
 219             canwrite = false;
 220             maybe_autoclose();
 221         }
 222
 223         public void maybe_autoclose()
 224         {
 225             if (!canread && !canwrite)
 226                 close();
 227         }
 228
 229         public virtual bool flush(int msec_timeout)
 230         {
 231             return true; // no buffer
 232         }
 233
 234         // WARNING: assumes the write() will succeed!  Use only on WvStreams
 235         // with a write buffer.
 236         public void print(string fmt, params object[] args)
 237         {
 238             print((object)wv.fmt(fmt, args));
 239         }
 240
 241         // WARNING: assumes the write() will succeed!  Use only on WvStreams
 242         // with a write buffer.
 243         public virtual void print(object o)
 244         {
 245             byte[] b = o.ToUTF8();
 246             int n = write(b);
 247             wv.assert(n == b.Length,
 248                       "Don't use print() on an unbuffered WvStream!");
 249         }
 250     }
 251
 252     // Wraps a WvStream in another WvStream, allowing us to override some
 253     // behaviour.  By default, a WvStreamClone just passes everything through
 254     // to the inner stream.
 255     public class WvStreamClone : WvStream
 256     {
 257         protected WvStream inner = null;
 258         bool hasinner { get { return inner != null; } }
 259
 260         public WvStreamClone(WvStream inner)
 261         {
 262             setinner(inner);
 263         }
 264
 265         public void setinner(WvStream inner)
 266         {
 267             if (inner != this.inner)
 268             {
 269                 if (hasinner)
 270                 {
 271                     this.inner.onreadable -= do_readable;
 272                     this.inner.onwritable -= do_writable;
 273                     this.inner.onclose -= do_close;
 274                 }
 275                 this.inner = inner;
 276                 if (hasinner)
 277                 {
 278                     if (can_onreadable) this.inner.onreadable += do_readable;
 279                     if (can_onwritable) this.inner.onwritable += do_writable;
 280                     this.inner.onclose += do_close;
 281                 }
 282             }
 283         }
 284
 285         public override bool ok {
 286             get { return base.ok && hasinner && inner.ok; }
 287         }
 288
 289         public override Exception err {
 290             get {
 291                 return hasinner ? inner.err : base.err;
 292             }
 293             set {
 294                 if (hasinner)
 295                     inner.err = value;
 296                 else
 297                     base.err = value;
 298             }
 299         }
 300
 301         public override EndPoint localaddr {
 302             get { return hasinner ? inner.localaddr : base.localaddr; }
 303         }
 304         public override EndPoint remoteaddr {
 305             get { return hasinner ? inner.localaddr : base.localaddr; }
 306         }
 307
 308         public override int read(WvBytes b)
 309         {
 310             if (hasinner)
 311                 return inner.read(b);
 312             else
 313                 return 0; // 0 bytes read
 314         }
 315
 316         public override int write(WvBytes b)
 317         {
 318             if (hasinner)
 319                 return inner.write(b);
 320             else
 321                 return 0; // 0 bytes written
 322         }
 323
 324         public override bool wait(int msec_timeout,
 325                                   bool readable, bool writable)
 326         {
 327             if (hasinner)
 328                 return inner.wait(msec_timeout, readable, writable);
 329             else
 330                 return base.wait(msec_timeout, readable, writable);
 331         }
 332
 333         public override void noread()
 334         {
 335             base.noread();
 336             if (hasinner)
 337                 inner.noread();
 338         }
 339
 340         public override void nowrite()
 341         {
 342             base.nowrite();
 343             if (hasinner)
 344                 inner.nowrite();
 345         }
 346
 347         public override bool flush(int msec_timeout)
 348         {
 349             if (hasinner)
 350                 return inner.flush(msec_timeout);
 351             else
 352                 return true;
 353         }
 354
 355         // We only want to register our callback with the inner stream if
 356         // we *have* a callback, and then only once.  Otherwise the stream
 357         // might start listening for read when we don't have any readable
 358         // handlers, resulting in it spinning forever.
 359         public override event Action onreadable {
 360             add { if (!can_onreadable) inner.onreadable += do_readable;
 361                   base.onreadable += value; }
 362             remove { base.onreadable -= value;
 363                      if (!can_onreadable) inner.onreadable -= do_readable; }
 364         }
 365         public override event Action onwritable {
 366             add { if (!can_onwritable) inner.onwritable += do_writable;
 367                   base.onwritable += value; }
 368             remove { base.onwritable -= value;
 369                      if (!can_onwritable) inner.onwritable -= do_writable; }
 370         }
 371
 372     }
 373
 374     /// Adds an input buffer to a WvStream.
 375     public class WvInBufStream : WvStreamClone
 376     {
 377         WvBuf inbuf = new WvBuf();
 378
 379         public WvInBufStream(WvStream inner) : base(inner)
 380         {
 381         }
 382
 383         public override int read(WvBytes b)
 384         {
 385             if (inbuf.used > 0)
 386             {
 387                 int max = inbuf.used > b.len ? b.len : inbuf.used;
 388                 b.put(0, inbuf.get(max));
 389                 post_readable();
 390                 return max;
 391             }
 392             else
 393                 return base.read(b);
 394         }
 395
 396         public string _getline(char splitchar)
 397         {
 398             if (inbuf.strchr(splitchar) > 0)
 399             {
 400                 post_readable(); // not stalled yet!
 401                 return inbuf.get(inbuf.strchr(splitchar)).FromUTF8();
 402             }
 403             return null;
 404         }
 405
 406         public string getline(int msec_timeout, char splitchar)
 407         {
 408             string l = _getline(splitchar);
 409             if (l != null) return l;
 410
 411             foreach (var remain in wv.until(msec_timeout))
 412             {
 413                 if (inner.wait(remain, true, false))
 414                     inner.read(inbuf, 4096);
 415
 416                 l = _getline(splitchar);
 417                 if (l != null) return l;
 418
 419                 if (inbuf.used == 0 && !ok)
 420                     break;
 421             }
 422
 423             if (!ok && inbuf.used > 0)
 424                 return inbuf.getall().FromUTF8();
 425
 426             return null;
 427         }
 428     }
 429
 430     /// Adds an output buffer to a WvStream
 431     public class WvOutBufStream : WvStreamClone
 432     {
 433         WvBuf outbuf = new WvBuf();
 434         bool writereg = true;
 435
 436         public WvOutBufStream(WvStream inner) : base(inner)
 437         {
 438         }
 439
 440         public override int write(WvBytes b)
 441         {
 442             outbuf.put(b);
 443             flush(0);
 444
 445             // always succeed
 446             return b.len;
 447         }
 448
 449         void _flush()
 450         {
 451             int wrote = base.write(outbuf.peekall());
 452             outbuf.get(wrote);
 453
 454             if (outbuf.used > 0 && !writereg)
 455             {
 456                 inner.onwritable += _flush;
 457                 writereg = true;
 458             }
 459             else if (outbuf.used == 0 && writereg)
 460             {
 461                 inner.onwritable -= _flush;
 462                 writereg = false;
 463             }
 464
 465             if (outbuf.used == 0)
 466                 post_writable();
 467         }
 468
 469         public override bool flush(int msec_timeout)
 470         {
 471             foreach (int remain in wv.until(msec_timeout))
 472             {
 473                 _flush();
 474                 if (remain != 0 && outbuf.used > 0)
 475                     wait(remain, false, true);
 476                 if (outbuf.used == 0)
 477                     return true;
 478             }
 479             return false;
 480         }
 481     }
 482
 483     /**
 484      * A combination of WvInBufStream and WvOutBufStream.
 485      *
 486      * We put the OutBufStream on the inside and the InBufStream on the
 487      * outside, because InBufStream actually adds API functions while
 488      * OutBufStream doesn't.  So getline() will work as expected on this
 489      * kind of stream.
 490      */
 491     public class WvBufStream : WvInBufStream
 492     {
 493         public WvBufStream(WvStream inner)
 494             : base(new WvOutBufStream(inner))
 495         {
 496         }
 497     }
 498 }