pipev.d

   1 /* ZX Spectrum Emulator
   2  * Copyright (C) 2017 Ketmar // Invisible Vector
   3  *
   4  * This program is free software: you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation, either version 3 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16  */
  17 module pipev;
  18
  19 import arsd.simpledisplay;
  20
  21
  22 public final class PipeEvent : Timer {
  23 private:
  24   int[2] fd = -1; // [0]: read; [1]: write
  25   void delegate (ubyte eid) onPulseCB;
  26
  27   int eventCount;
  28
  29 public:
  30   this (void delegate (ubyte eid) onPulse) {
  31     import core.sys.posix.unistd : pipe;
  32     if (pipe(fd) != 0) throw new Exception("can't create pipe event");
  33     onPulseCB = onPulse;
  34     super(int.max/8, delegate () {}, fd[0]);
  35   }
  36
  37   ~this () { close(); }
  38
  39   protected override void trigger () {
  40     //conwriteln("!!!");
  41     auto id = recv();
  42     if (onPulseCB !is null) onPulseCB(id);
  43   }
  44
  45   @property bool isOpen () const pure nothrow @safe @nogc { return (fd[0] >= 0); }
  46
  47   void close () /*nothrow @trusted @nogc*/ {
  48     destroy();
  49     import core.sys.posix.unistd : close;
  50     if (fd[0] >= 0) { close(fd[0]); fd[0] = -1; }
  51     if (fd[1] >= 0) { close(fd[1]); fd[1] = -1; }
  52   }
  53
  54   void send (ubyte id/*, const(ubyte)[] data=null*/) {
  55     {
  56       import core.atomic;
  57       atomicOp!"+="(*cast(shared int*)&eventCount, 1);
  58     }
  59     /*synchronized(this)*/ {
  60       if (!isOpen) throw new Exception("can't send event to closed pipe");
  61       if (!rawWrite(fd[1], (&id)[0..1])) throw new Exception("pipe writing error");
  62     }
  63     /*
  64     size_t len = data.length;
  65     if (!rawWrite(fd[1], (&len)[0..1])) throw new Exception("pipe writing error");
  66     if (!rawWrite(fd[1], data)) throw new Exception("pipe writing error");
  67     */
  68   }
  69
  70   ubyte recv () {
  71     /*synchronized(this)*/ {
  72       if (!isOpen) throw new Exception("can't receive event from closed pipe");
  73       ubyte[1] id;
  74       if (!rawRead(fd[0], id[])) throw new Exception("pipe reading error");
  75       {
  76         import core.atomic;
  77         atomicOp!"-="(*cast(shared int*)&eventCount, 1);
  78       }
  79       return id[0];
  80     }
  81   }
  82
  83   bool hasSomething () {
  84     {
  85       import core.atomic;
  86       return (atomicLoad(*cast(shared int*)&eventCount) != 0);
  87     }
  88     /*
  89     synchronized(this) {
  90       if (!isOpen) return false; // obviously
  91       import core.sys.posix.sys.select;
  92       import core.sys.posix.sys.time;
  93       fd_set rd;
  94       timeval tv;
  95       for (;;) {
  96         FD_ZERO(&rd);
  97         FD_SET(fd[0], &rd);
  98         tv.tv_sec = 0;
  99         tv.tv_usec = 0;
 100         auto res = select(fd[0]+1, &rd, null, null, &tv);
 101         if (res < 0) {
 102           import core.stdc.errno : errno, EINTR;
 103           if (errno != EINTR) return false;
 104         } else {
 105           return (res > 0);
 106         }
 107       }
 108     }
 109     */
 110   }
 111
 112 static final:
 113   bool rawWrite (int fd, const(void)[] data) {
 114     import core.sys.posix.unistd : write;
 115     if (fd < 0) return false;
 116     auto bp = cast(const(ubyte)*)data.ptr;
 117     auto left = data.length;
 118     while (left > 0) {
 119       auto wr = write(fd, bp, left);
 120       if (wr < 0) {
 121         import core.stdc.errno : errno, EINTR;
 122         if (errno != EINTR) return false;
 123       } else if (wr == 0) {
 124         return false;
 125       } else {
 126         bp += wr;
 127         left -= wr;
 128       }
 129     }
 130     return true;
 131   }
 132
 133   bool rawRead (int fd, void[] data) {
 134     import core.sys.posix.unistd : read;
 135     if (fd < 0) return false;
 136     auto bp = cast(ubyte*)data.ptr;
 137     auto left = data.length;
 138     while (left > 0) {
 139       auto rd = read(fd, bp, left);
 140       if (rd < 0) {
 141         import core.stdc.errno : errno, EINTR;
 142         if (errno != EINTR) return false;
 143       } else if (rd == 0) {
 144         return false;
 145       } else {
 146         bp += rd;
 147         left -= rd;
 148       }
 149     }
 150     return true;
 151   }
 152 }