From e5b2f7c6054c419d8a031e8825fbebae1c166a19 Mon Sep 17 00:00:00 2001
From: sbosse <sbosse@uni-bremen.de>
Date: Mon, 21 Jul 2025 22:47:38 +0200
Subject: [PATCH] Mon 21 Jul 22:43:21 CEST 2025

---
 js/dos/connFIFOsrv.js | 621 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 621 insertions(+)
 create mode 100644 js/dos/connFIFOsrv.js

diff --git a/js/dos/connFIFOsrv.js b/js/dos/connFIFOsrv.js
new file mode 100644
index 0000000..73a45ee
--- /dev/null
+++ b/js/dos/connFIFOsrv.js
@@ -0,0 +1,621 @@
+/**
+ **      ==============================
+ **       O           O      O   OOOO
+ **       O           O     O O  O   O
+ **       O           O     O O  O   O
+ **       OOOO   OOOO O     OOO  OOOO
+ **       O   O       O    O   O O   O
+ **       O   O       O    O   O O   O
+ **       OOOO        OOOO O   O OOOO
+ **      ==============================
+ **      Dr. Stefan Bosse http://www.bsslab.de
+ **
+ **      COPYRIGHT: THIS SOFTWARE, EXECUTABLE AND SOURCE CODE IS OWNED
+ **                 BY THE AUTHOR(S).
+ **                 THIS SOURCE CODE MAY NOT BE COPIED, EXTRACTED,
+ **                 MODIFIED, OR OTHERWISE USED IN A CONTEXT
+ **                 OUTSIDE OF THE SOFTWARE SYSTEM.
+ **
+ **    $AUTHORS:     Stefan Bosse
+ **    $INITIAL:     (C) 2006-2017 bLAB
+ **    $CREATED:     31-10-16 by sbosse.
+ **    $VERSION:     1.2.2
+ **
+ **    $INFO:
+ **
+ **  ==============================================
+ **  DOS: Broker Connection Module 
+ **  Server, FS FIFO communication, limited to 2 connections (node.js bug)
+ **  Data transfer: JSON
+ **  ==============================================
+ *
+ *
+ * Two unidirectional FIFOs are used for bidirectional client-server communication.
+ * There is one shared public channel request link, and multiple private P2P client-server links (FIFO pair).
+ * Clients writing initially to the request channel to get a free private channel from the server.
+ * Note: All clients simultaneously requesting a channel will receive multiple replies for different clients!
+ * A FIFO channel is shared by all participants!
+ *
+ * 
+ *
+ * Notes:
+ * Superfluous? All http function callback computations are wrapped by scheduler callbacks 
+ * to avoid main thread preemption!
+ * Is event queuing always guaranteed in JS?
+ *
+ *
+ *
+ **    $ENDOFINFO
+ */
+"use strict";
+var log = 0;
+
+var Io = Require('com/io');
+var Net = Require('dos/network');
+var Buf = Require('dos/buf');
+var Rpc = Require('dos/rpc');
+var util = Require('util');
+var net = Require('net');
+var Sch = Require('dos/scheduler');
+var Comp = Require('com/compat');
+var Perv = Comp.pervasives;
+var String = Comp.string;
+var Array = Comp.array;
+var Filename = Comp.filename;
+var trace = Io.tracing;
+var div = Perv.div;
+var Conn = Require('dos/connutils');
+var Fs = Require('fs');
+
+var isNode = Comp.isNodeJS();
+
+var Mode = Conn.Mode;
+
+
+/**************************
+ ** FIFO Broker SERVER Interface
+ **************************/
+/**
+ *
+ * @param {port} hostport
+ * @param {string} srv_url
+ * @param {string} srv_ipport
+ * @param {rpcrouter} router
+ * @constructor
+ */
+// tyepof options = {hostport,path,channels,router,verbose,env,mode} 
+// typeof mode = 'unidir'|'bidir'
+var Server = function(options) {
+  this.env=options.env||{};
+  this.hostport=options.hostport;               // Public communication Net.Port == Host port
+  this.status = Net.Status.STD_OK;
+  this.router=options.router;
+  this.app=undefined;
+  this.server=undefined;
+  this.client_queue=[];
+  this.hosts={};
+  this.lock=false;
+  this.mode=Mode.TWOCHAN;   
+  this.keepalive=options.keepalive||true;
+  
+  // FIFO file system path
+  this.path=options.path||'/tmp/broker';
+  // Number of channels
+  this.channels=Math.min(options.channels||2,2);
+  
+  this.conn_port = Net.uniqport();
+  this.rpccon=Rpc.RpcConn(
+      self.conn_port,
+      /*
+      ** Connection Forward and Deliver Operation
+      ** ONECHAN client connectivity: collect
+      ** TWOCHAN client connectivity: forward
+      */
+      function(rpcio,callback) {
+        // TODO rpcio -> msg
+      },
+      // We're handling multiple client connections - we're alive
+      function() {
+          return true;
+      }
+  );
+  this.verbose=options.verbose||0;
+  this.enabled=false;
+  this.warned=0;
+  this.last=undefined;
+  
+  // Transfer of multi-part messages?
+  this.multipart=true;
+};
+
+/** Client-App request (collecting pending and queued TRANSREQ/TRANSREP/LOOKUP messages)
+ *  Used only in ONECHAN mode!! 
+ */
+
+Server.prototype.collect = function (self, msg) {   // HTTP Callback handler
+  var res,
+      timeout=(msg.timeout != undefined) ? (Perv.int_of_string(msg.timeout)) : 0;
+  if (timeout) msg.timeout=timeout;
+  
+  this.rpccon.stats.op_forward++;
+  /*
+   ** 1. Forward TRANS message requests if there are pending requests for the host port.
+   ** 2. Forward TRANS message replies if there are pending replies for the host port.
+   ** 3. Forward LOOKUP messages (broadcast)
+   ** If there are actually no messages, add this request to the request queue (if timeout <>0).
+   */
+  /*
+  ** Forward all pending transaction requests and replies...
+   */
+  Io.log(((log+self.verbose) < 2) || ('[BFIS] collect for host ' + Net.Print.port(msg.hostport)+
+      ' #trans='+ Array.length(self.router.trans_queue)+' #lookup='+Array.length(self.router.lookup_queue)));
+  res=this.schedule_client(msg,true);
+
+  if (res && msg.timeout)
+    self.client_queue.push(msg);
+};
+
+/** Client-App forwarding (send pending and queued TRANSREQ/TRANSREP/LOOKUP messages)
+ *  Used only in TWOCHAN mode!! Called from router.
+ */
+
+Server.prototype.forward = function (msg) {   // HTTP Callback handler
+  var res=0,
+      self=this;
+  /*
+   ** 1. Forward TRANS message request if there are pending requests for the host port.
+   ** 2. Forward TRANS message reply if there are pending replies for the host port.
+   ** 3. Forward LOOKUP messages (broadcast)
+   */
+  /*
+  ** Forward all pending transaction requests and replies...
+   */
+  Io.log(((log+self.verbose) < 2) || ('[BFIS] forward for host ' + Net.Print.port(msg.hostport)+
+      ' #trans='+ Array.length(self.router.trans_queue)+' #lookup='+Array.length(self.router.lookup_queue)));
+
+  if (msg.status==undefined) msg.status='EOK';
+
+  // console.log(msg)
+  if (this.hosts[msg.sendport]) {
+    res++;
+    self.send(self.reply(this.hosts[msg.sendport],msg));
+  }
+
+  if (res) this.rpccon.stats.op_forward++;
+
+  return res;
+};
+
+/** Initialize this server
+ *
+ */
+Server.prototype.init = function () {
+  var self = this,pathI,pathO;
+  // Service client requests
+  this.enabled=true;
+  function receiver (i) {
+      var chI, chO,
+          pathI=self.path+i+'I',
+          pathO=self.path+i+'O';
+      if (!Fs.existsSync(pathO)) {
+        Io.out('[BFIS] Cannot open ' + pathO+': not existing!');          
+        return;
+      }
+      if (!Fs.existsSync(pathI)) {
+        Io.out('[BFIS] Cannot open ' + pathI+': not existing!');          
+        return;
+      }
+      chO=Fs.createWriteStream(pathO);
+
+      function listen () {
+        // Wait for a client connecting to the FIFO
+        Io.out('[BFIS] Listening on ' + pathI+' in mode '+self.mode+' '+(self.keepalive?'KEEPALIVE':''));
+        chI=Fs.createReadStream(pathI);
+        chI.on('open',function () {
+          Io.log(((log+self.verbose)<2)||('[BFIS] Connected on ' + pathI));
+        });
+        chI.on('close',function () {
+          Io.log(((log+self.verbose)<2)||('[BFIS] Disconnect on ' + pathI));
+          if (self.enabled) listen();
+        });
+        chI.on('error',function () {
+          Io.log(((log+self.verbose)<2)||('[BFIS] Disconnect on ' + pathI));
+          if (self.enabled) listen();
+        });
+        chI.on('data',function (chunk) {
+          var req,part,parts,data;
+          Io.log(((log+self.verbose)<2)||('[BFIS] Received: ' + chunk.length));
+          data=chunk.toString('ascii', 0, chunk.length);
+          Io.log(((log+self.verbose)<2)||('[BFIS] Received: ' + data));
+          parts=Conn.splitData(data);
+          if (self.last) parts[0]=self.last+parts[0];
+          if (Array.last(parts) != '') self.last=Array.last(parts); else self.last=undefined;
+
+          // console.log(parts)
+
+          for(part in parts) {
+            if (parts[part]=='') continue;
+            req = JSON.parse(parts[part]);
+            if (self.multipart) Conn.decode(req);
+            req.socket=chO;
+            self.request(self,req);
+          }              
+        });
+      }
+      listen();
+  }
+  this.server=[];
+  for (var i=0;i<this.channels;i++) {
+    pathI=self.path+i+'I';
+    pathO=self.path+i+'O';
+    if (!Fs.existsSync(pathI)) {
+       try {
+        Io.execSync('mkfifo '+  pathI);
+        Io.out('[BFIS] Created FIFO channel ' + pathI);        
+       } catch (e) {};
+    };
+    if (!Fs.existsSync(pathO)) {
+       try {
+        Io.execSync('mkfifo '+  pathO);
+        Io.out('[BFIS] Created FIFO channel ' + pathO);        
+       } catch (e) {};
+    };
+  
+    try {receiver(i);} catch (e) {Io.out('[BFIS] Error: '+e)};
+  }
+  this.router.add_conn(self.rpccon);
+};
+
+Server.prototype.log = function (v) {log=v};
+
+
+/** Create a reply message from a request message and a reply object.
+ *
+ */
+Server.prototype.reply = function (msg,obj) {
+  var reply={},
+      p;
+  reply.socket=msg.socket;
+  
+  for (p in obj) reply[p]=obj[p];
+  if (msg.tid != undefined) reply.tid=msg.tid;
+  if (msg.rid != undefined) reply.rid=msg.rid;
+  
+  return reply;
+}
+
+/** Main request entry point.
+ *
+ */
+Server.prototype.request = function (self, msg) {
+  
+  var i,str,
+       buf, rpcio, hdr;
+  this.rpccon.stats.op_get++;
+  Io.log(((log+self.verbose) < 2) || ('[BFIS] server request: ' + (msg.rpc||msg.type)+' from '+msg.ip+':'+msg.ipport));
+  Io.log(((log+self.verbose) < 3) || ('[BFIS] server request: ' + util.inspect(msg)));
+  // Io.trace(trace || ('[BFIS] server request: ' + req.url));
+
+  if (this.multipart) Conn.decode(msg);
+
+  if (msg.rpc != undefined &&
+      msg.rpc == 'request' &&
+      msg.hostport != undefined)
+  {
+      /**************************************************
+       ** RPC forward request (Client- /Browser App. collector request)
+       *************************************************/
+      Sch.ScheduleCallback([Sch.Bind(self,self.collect),self, msg]);
+  } else if (msg.rpc != undefined &&
+      msg.rpc == 'iamhere' &&
+      msg.hostport != undefined &&
+      msg.srvport != undefined)
+  {
+      /********************************
+       ** short IAMHERE rpc message
+       ********************************/
+      {
+          if (self.verbose && self.router.lookup_host(msg.hostport)==undefined) 
+            Io.out('[BFIS] IAMHERE! Adding remote host ' + 
+               Net.Print.port(msg.hostport) + ' ip=' + req.connection.remoteAddress + 
+              ((msg.port && msg.port!='undefined')?(' ipport=' + msg.port):''));
+          self.router.add_host(msg.hostport, self.rpccon.port);
+          var rpcio=self.router.pkt_get();
+          rpcio.operation=Rpc.Operation.IAMHERE;
+          rpcio.connport=self.rpccon.port;
+          rpcio.hostport=msg.hostport;
+          rpcio.sendport=self.router.hostport;
+          rpcio.header.h_port=msg.srvport;
+          self.router.route(rpcio);
+
+      }
+      /*
+      {
+          self.router.add_port(srvport, hostport);
+          self.router.add_host(hostport, self.rpccon.port);
+          Io.log((log < 1) || ('[BFIS] server get IAMHERE: adding server port ' +
+              Net.Print.port(srvport) + ' located on host ' + Net.Print.port(hostport)));
+          self.router.remove_lookup(srvport);
+          // TBD: search for stalled transactions that must be send to another server capable host!
+          // Broker connected client apps. will collect the appropriate transaction on the next request round.
+          //res.
+      }
+      */
+      
+      self.send(self.reply(msg,{status:'EOK'}));
+  } else if (msg.rpc != undefined &&
+             msg.rpc == 'trans' &&
+             msg.hostport != undefined &&
+             msg.tid != undefined)
+  {  
+      /**************************************************
+       ** RPC TRANS message (Client- /Browser App. transaction request)
+       *************************************************/
+      
+      rpcio = self.router.pkt_get();
+      buf = Buf.Buffer(msg.data.header);
+      Buf.buf_get_hdr(buf, rpcio.header); 
+      Io.log((log < 2) || (('[BFIS] trans header: ' + Net.Print.header(rpcio.header))));
+      rpcio.init(Rpc.Operation.TRANSREQ, rpcio.header, msg.data.data);
+      rpcio.context=undefined;  // rpc not from here!
+      rpcio.tid = msg.tid;
+      rpcio.hostport = msg.hostport;
+      //rpcio.socket = msg.socket;
+      
+      self.hosts[msg.hostport]={socket:msg.socket,rid:msg.rid,hostport:msg.hostport,timeout:5000};
+      // if (local request) service ..
+      self.router.add_host(rpcio.hostport,self.rpccon.port);
+      self.router.route(rpcio);
+      // TBD: Check for local waiting servers in routing_db...
+      //res.
+  
+  } else if (msg.rpc != undefined &&
+             msg.rpc == 'reply' &&
+             msg.hostport != undefined &&
+             msg.tid != undefined) {
+      /*************************
+       * RPC Transaction Reply
+       ************************/
+  
+      /*
+      ** Reply for a transaction must be forwarded to the transaction origin collected by the source host application and
+      ** the next http get/?rpc=request request.
+      */
+      // New transaction request from client app., add it to the transaction queue.
+      buf = Buf.Buffer(msg.data.header);
+      rpcio = self.router.pkt_get();
+      Buf.buf_get_hdr(buf,rpcio.header);
+      rpcio.init(Rpc.Operation.TRANSREP, rpcio.header, msg.data.data);
+      rpcio.context=undefined;    // rpc not from here!
+
+      rpcio.tid = msg.tid;
+      rpcio.hostport = msg.hostport;
+      rpcio.sendport = msg.sendport;
+
+      Io.log(((log+self.verbose) < 2)||('[BFIS] post: got reply, hostport=' +
+              Net.Print.port(msg.hostport) + ' tid=' + tid + ' srvport=' + Net.Print.port(rpcio.header.h_port)));
+      self.router.route(rpcio);
+      //res.
+      self.send(self.reply(msg,{status:'EOK'}));
+  
+  } else if (msg.type != undefined &&
+             msg.type == 'alive' &&
+             msg.hostport != undefined)
+  {
+      /***************************
+       ** ALIVE message request
+       ***************************/
+      this.rpccon.stats.op_alive++;
+
+      buf=Buf.Buffer();
+
+      if (self.verbose && self.router.lookup_host(msg.hostport)==undefined) 
+        Io.out('[BFIS] ALIVE! Adding remote host ' + 
+               Net.Print.port(msg.hostport) + ' ip=' + msg.ip + (msg.ipport!='undefined'?(' ipport=' + msg.ipport):''));
+      // TODO this.router.add_host(hostport, [msg.url, msg.port]);
+      self.router.add_host(msg.hostport,self.rpccon.port);
+      self.hosts[msg.hostport]={socket:msg.scoket,hostport:msg.hostport,timeout:5000};
+
+      //res.
+      Buf.buf_put_port(buf,self.hostport);
+      self.send(self.reply(msg,{status:'EOK',data:Buf.buf_to_hex(buf)}));
+  } else if (msg.type != undefined &&
+             msg.type == 'ask' &&
+             msg.xname != undefined)
+  {
+      /***************************
+       ** ASK message request
+       ***************************/
+      this.rpccon.stats.op_ask++;
+      buf=Buf.Buffer();
+      //console.log(self.env[msg.xname]);
+      Buf.buf_put_string(buf,self.env[msg.xname]||'undefined');
+      self.send(self.reply(msg,{status:'EOK', data:Buf.buf_to_hex(buf)}));
+  } else if (msg.type != undefined &&
+             msg.type == 'notify' &&
+             msg.xname != undefined &&
+             msg.xvalue !=undefined)
+  {
+      /***************************
+       ** NOTIFY message request
+       ***************************/
+      this.rpccon.stats.op_notify++;
+      self.env[msg.xname]=msg.xvalue;
+      //console.log(self.env[msg.xname]);
+      self.send(self.reply(msg,{status:'EOK'}));
+  }    // res.
+  else
+      self.send(self.reply(msg,{status:'EINVALID'}));
+};
+
+
+
+/** Schedule pending transaction replies for a client (message)
+ *
+ */
+Server.prototype.schedule_client = function (client /* msg */, discard) {
+  var msgn = 0,
+      self = this,
+      data = [],
+      buf, trans, cache, lookup;
+
+
+  Io.log((self.router.monitor < 1) || ('[BFIS] schedule_client for host ' + Net.Print.port(client.hostport)+
+      ' #trans='+ Array.length(self.router.trans_queue)+' #lookup='+Array.length(self.router.lookup_queue)));
+  do {
+      trans = self.router.lookup_trans_for_host(client.hostport);
+      Io.log(((log+self.verbose)<10)||('[BFIS] schedule_client: lookup_trans_for_host=' + util.inspect(trans)));
+      if (trans != undefined) {
+          Io.log(((log+self.verbose)<2)||('[BFIS] schedule_client: found trans=' + Rpc.Print.rpcio(trans)));
+
+          if (trans.operation == Rpc.Operation.TRANSREQ) {
+              data.push(trans.to_json('extended'));
+              msgn++;
+              if (discard) self.router.pkt_discard(trans);
+          } else if (trans.operation == Rpc.Operation.TRANSREP) {
+              // Transaction reply!
+              data.push(trans.to_json('extended'));
+              msgn++;
+              if (discard) self.router.pkt_discard(trans);
+          }
+      }
+  } while (trans != undefined);
+  /*
+  ** Add all pending server port LOOKUP requests...
+  ** Forward a LOOKUP of a server port search only once to a client host!
+  */
+  for (var i in self.router.lookup_queue) {
+      lookup = self.router.lookup_queue[i];
+      // TODO: should we retransmit lookup requests here?
+      cache = self.router.lookup_cache[i];
+      if (cache[client.hostport]==undefined) {
+          cache[client.hostport] = 1;
+          buf = Buf.Buffer();
+          Buf.buf_put_port(buf, lookup.header.h_port);
+          data.push({
+            rpc:'lookup',
+            hostport:self.hostport,
+            data:Buf.buf_to_hex(buf)
+          });
+          msgn++;
+      }
+  }
+  if (msgn>0) {
+      self.send(self.reply(client,{status:'EOK',data:data}));
+      return false;
+  } else if (client.timeout<=0) {
+      self.send(self.reply(client,{status:'ENOENTR'}));
+      return false;
+  } else return true;
+}
+
+/*
+**  Schedule all pending client-side requests if there is matching input.
+**  Called from router function. Lock required!
+*/
+
+Server.prototype.schedule = function () {
+  var self = this;
+  
+  if (self.lock) return;
+  this.rpccon.stats.op_schedule++;
+  Io.log(((log+self.verbose)<2) || ('[BFIS] server schedule ['+this.client_queue.length+'] .. '));
+  Io.trace(trace || ('[BFIS] server schedule ['+this.client_queue.length+'] .. '));
+  self.lock = true;
+  self.client_queue = Array.filter(self.client_queue, function (client) {
+      /*
+       * Forward all matching pending transaction requests and replies...
+       */
+      Io.log(((log+self.verbose) < 2) || ('[BFIS] schedule for client host ' + Net.Print.port(client.hostport)+
+          ' #trans='+ Array.length(self.router.trans_queue)+' #lookup='+Array.length(self.router.lookup_queue)));
+      return self.schedule_client(client,true);
+  });
+  self.lock = false;
+};
+
+/** Send a message to a destination
+ *
+ */
+// function send(repsonse:socket,data:string)
+Server.prototype.send = function (msg) {
+  var client, 
+      self=this,
+      socket,
+      data;
+  //response.writeHead(200, {'Content-Type': 'text/plain; charset=utf8'});
+  //response.writeHead(200);
+  this.rpccon.stats.op_send++;
+  socket=msg.socket; msg.socket=undefined;
+  Conn.encode(msg);
+  
+  Io.log(((log+this.verbose)<2)||('[BFIS] send ['+util.inspect(msg)+'] -> '+socket));
+  if (socket) {
+    data = JSON.stringify(msg);
+// console.log(data);
+    if (this.multipart) socket.write(data+Conn.EOM);
+    else socket.write(data); 
+    this.warned=0;
+  };
+};
+
+
+
+
+
+/**
+ *
+ * @param {number} [interval]
+ */
+Server.prototype.start = function (interval) {
+  var self = this;
+  if (!interval) interval=1000;
+  /*
+  ** Start a client service and garbage collector for blocked client-side transactions (self.client_queue)
+   */
+  Sch.AddTimer(interval, 'Broker Client Service and Garbage Collector', function (context) {
+      var p;
+      Io.log(((log+self.verbose)<4)||('[BFIS] Client Service and Garbage Collector'));
+      if (!self.lock) {
+          self.lock = true;
+
+          if (self.mode==Mode.ONECHAN)
+            self.client_queue = Array.filter(self.client_queue, function (client /* msg */) {
+              client.timeout=client.timeout-interval;              
+              return self.schedule_client(client,true);
+            });
+          else {
+            for (p in self.hosts) {
+              // console.log(self.hosts[p])
+              if (self.hosts[p]) {
+                self.hosts[p].timeout = self.hosts[p].timeout - interval;
+                if (self.hosts[p].timeout <= 0) {
+                  Io.log(((log+self.verbose)<1)||('[BFIS] Host cache: Removing host '+
+                          ' hostport='+Net.Print.port(self.hosts[p].hostport)));
+                  self.hosts[p]=undefined;
+                }
+              }
+            }
+          } 
+            
+          self.lock = false;
+      } else {
+          context.blocked=false;
+          context.timer=Sch.time;
+          Sch.ScheduleNext();
+      }
+  });
+};
+
+
+module.exports = {
+  /** RCP Broker Server for client-capable-only applications
+   *
+   */
+  // typeof options = {hostport:port,srv_ip,srv_ipport,router,verbose?,env?:{}}
+  Mode: Mode,
+  Server: function(options) {
+      var obj = new Server(options);
+      Object.preventExtensions(obj);
+      return obj;
+  }
+};