text: /* stomp.c  -  STOMP 1.1 protocol for HIIOS engine
   * Copyright (c) 2012 Sampo Kellomaki (sampo@iki.fi), All Rights Reserved.
   * This is confidential unpublished proprietary source code of the author.
   * NO WARRANTY, not even implied warranties. Contains trade secrets.
   * Distribution prohibited unless authorized in writing. See file COPYING.
   * Special grant: http.c may be used with zxid open source project under
   * same licensing terms as zxid itself.
   * $Id$
   *
   * 16.8.2012, created, based on http and smtp --Sampo
   * 19.8.2012, added tolerance for CRLF where strictly LF is meant --Sampo
   *
   * STOMP 1.1 frame is considered a PDU and consists of
   *
   *   \n              -- zero, or more in case of heart beats
   *   COMMAND\n       -- lines end in LF (not CRLF)
   *   header:value\n  -- zero or more (content-length is one possible header)
   *   \n              -- blank line separates headers and body
   *   payload
   *   \0
   *   \n              -- zero, or more in case of heart beats
   *
   * See also:  http://stomp.github.com/stomp-specification-1.1.html (20110331)
   *            http://stomp.github.io/
   * Todo: implement heart beat generation and checking
   */
  
  #include "platform.h"
  #include "errmac.h"
  #include "akbox.h"
  #include "hiios.h"
  #include "hiproto.h"
  #include <zx/zxidconf.h>
  #include <zx/zxidutil.h>
  
  #include <ctype.h>
  #include <memory.h>
  #include <stdlib.h>
  #include <netinet/in.h> /* htons(3) and friends */
  #include <sys/types.h>
  #include <sys/stat.h>
  #include <fcntl.h>
  #include <errno.h>
  
  /* Alias some struct fields for headers that can not be seen together. */
  #define receipt   host
  #define rcpt_id   host
  #define acpt_vers vers
  #define tx_id     vers
  #define session   login
  #define subs_id   login
  #define subsc     login
  #define server    pw
  #define ack       pw
  #define msg_id    pw
  #define heart_bt  dest
  #define zx_rcpt_sig dest
  
  extern int verbose;  /* defined in option parsing in zxbusd.c */
  extern zxid_conf* zx_cf;
  
  #if 0
  /* Called by: */
  static struct hi_pdu* stomp_encode_start(struct hi_thr* hit)
  {
    struct hi_pdu* resp = hi_pdu_alloc(hit,0,"stomp_enc_start");
    if (!resp) { hi_dump(hit->shf); NEVERNEVER("*** out of pdus in bad place %d", 0); }
    return resp;
  }
  #endif
  
  /*() Send ERROR to remote client. */
  
  /* Called by:  stomp_frame_err, stomp_got_login x2, zxbus_persist x2 */
  int stomp_err(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req, const char* ecode, const char* emsg)
  {
    int len;
    char* rcpt;
    if ((rcpt = req->ad.stomp.receipt)) {
      len = memchr(rcpt, '\n', req->ap - rcpt) - (void*)rcpt;
    } else {
      len = 1;
      rcpt = "-";
    }
    ERR("%s", emsg);
    hi_sendf(hit, io, 0, req, "ERROR\nmessage:%s\nreceipt-id:%.*s\ncontent-type:text/plain\ncontent-length:%d\n\n%s%c", ecode, len, rcpt, strlen(emsg), emsg, 0);
    return HI_CONN_CLOSE;
  }
  
  /*() Send an error early on in decode process */
  
  /* Called by:  stomp_decode x2 */
  static int stomp_frame_err(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req, const char* emsg)
  {
    /* At this early stage the req is still a io->cur_pdu. We need to
     * promote it to a real request so that the free logic will work right. */
    hi_add_to_reqs(hit, io, req, STOMP_MIN_PDU_SIZE);
    return stomp_err(hit,io,req,"malformed frame received",emsg);
  }
  
  #define CMD_NI_MSG "Command(%.*s) not implemented by server."
  
  /*() Send not implemented ERROR to remote client. */
  
  /* Called by:  stomp_decode x7, stomp_got_unsubsc */
  static int stomp_cmd_ni(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req, const char* cmd)
  {
    const char* nl = strchr(cmd, '\n');
    int len;
    char* rcpt;
    if ((rcpt = req->ad.stomp.receipt)) {
      len = memchr(rcpt, '\n', req->ap - rcpt) - (void*)rcpt;
    } else {
      len = 1;
      rcpt = "-";
    }
    ERR(CMD_NI_MSG, (int)(nl-cmd), cmd);
    hi_sendf(hit, io, 0, req, "ERROR\nmessage:command not implemented by server\nreceipt-id:%.*s\ncontent-type:text/plain\ncontent-length:%d\n\n" CMD_NI_MSG "%c", len, rcpt, sizeof(CMD_NI_MSG)-3+(nl-cmd), nl-cmd, cmd, 0);
    return HI_CONN_CLOSE;
  }
  
  /*() Got ERROR from remote client. */
  
  /* Called by:  stomp_decode */
  static int stomp_got_err(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    /*struct hi_pdu* resp = stomp_encode_start(hit);*/
    /*hi_sendv(hit, io, 0, req, resp, len, resp->m, size, req->m + len);*/
    ERR("remote sent error(%.*s)", (int)(req->ap-req->m), req->m);
    return HI_CONN_CLOSE;
  }
  
  /*() Send a receipt to client. */
  
  /* Called by:  stomp_got_disc, stomp_got_send, stomp_got_zxctl, zxbus_subscribe */
  void stomp_send_receipt(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    int len;
    char* rcpt;
    char sigbuf[1024];
    if ((rcpt = req->ad.stomp.receipt)) {
      len = (char*)memchr(rcpt, '\n', req->ap - rcpt) - rcpt;
    } else {
      len = 1;
      rcpt = "-";
    }
    DD("rcpt(%.*s) len=%d", len, rcpt, len);
  
    zxbus_mint_receipt(zx_cf, sizeof(sigbuf), sigbuf,
  		     len, rcpt,
  		     -2, req->ad.stomp.dest,
  		     -1, io->ent->eid,   /* entity to which we issue receipt */
  		     req->ad.stomp.len, req->ad.stomp.body);
    hi_sendf(hit, io, 0, req, "RECEIPT\nreceipt-id:%.*s\nzx-rcpt-sig:%s\n\n%c", len, rcpt, sigbuf,0);
  }
  
  /* STOMP Received Command Handling */
  
  /* Called by:  stomp_decode */
  static int stomp_got_login(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    if (!req->ad.stomp.login)
      return stomp_err(hit, io, req, "login fail", "No login header supplied (client error). zxbusd(8) requires login header whose value is the EntityID of the connecting client.");
    
    if (zxbus_login_ent(hit, io, req)) {
      hi_sendf(hit, io, 0, req, "CONNECTED\nversion:1.1\nserver:zxbusd-1.x\n\n%c", 0);
      return 0;
    } else
      return stomp_err(hit, io, req, "login fail", "login failed either due to nonexistent entity id or bad credential");
  }
  
  /* Called by:  stomp_decode */
  static int stomp_got_disc(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    /* *** figure out the real receipt id */
    stomp_send_receipt(hit, io, req);
    return HI_CONN_CLOSE;
  }
  
  /*() Main function for receiving audit bus summary log lines from SPs.
   * This function will first store the line in a persistent way (RAID1
   * arrangement needs to be implemented at the OS level), and then,
   * perhaps, attempt to deliver the message to all subscribers. Indeed,
   * delivery should be attempted first and if successful, the persistence
   * is not necessary. If delivery is unsuccessful, the delivery needs
   * to be retried, i.e. this is a store-and-forward system. However,
   * the delivery first approach needs extensive IO engine
   * operations and it thus may be easier to just store the message
   * first and then have a separate process attempt the sending. This
   * latter is the approach adopted here. */
  
  /* Called by:  stomp_decode */
  static void stomp_got_send(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    if (zxbus_persist(hit, io, req)) {
      stomp_send_receipt(hit, io, req);
    } else {
      ERR("Persist Problem. Disk full? %d", 0);
      //hi_sendf(hit, io, 0, req, "ERROR\nmessage:persist failure\nreceipt-id:%.*s\n\nUnable to persist message. Can not guarantee reliable delivery, therefore rejecting.%c", len, rcpt, 0);
    }
  }
  
  /*() Find a request that matches response. Looks in
   * the io->pending list for message ID match. When found,
   * the req is dequeued from pending. */
  
  /* Called by:  stomp_got_ack, stomp_got_nack */
  static struct hi_pdu* stomp_find_pending_req_for_resp(struct hi_io* io, struct hi_pdu* resp)
  {
    struct hi_pdu* prev;
    struct hi_pdu* req;
    int midlen=resp->ad.stomp.msg_id?(strchr(resp->ad.stomp.msg_id,'\n')- resp->ad.stomp.msg_id):0;
    
    LOCK(io->qel.mut, "ack");
    for (prev = 0, req = io->pending; req; prev = req, req = req->n) {
      if (!memcmp(resp->ad.stomp.msg_id, req->ad.stomp.msg_id, midlen+1)) {
        if (prev)
  	prev->n = req->n;
        else
  	io->pending = req->n;
        resp->req = req;
        resp->parent = req->parent;
        break;
      }
    }
    UNLOCK(io->qel.mut, "ack");
    return req;
  }
  
  /*() Process NACK response from client to MESSAGE request sent by server.
   * This is essentially nice way for the client to communicate to us it has
   * difficulty in persisting the message. It could also just hang up and the
   * net effect would be the same. However, receiving the NACK allows us to
   * close the delivery batch sooner so we can free the memory in the
   * hopeless cases quicker. (*** there should also be a handler for
   * close-connection lost that would do similar cleanup) */
  
  /* Called by:  stomp_decode */
  static void stomp_got_nack(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* resp)
  {
    int sublen, midlen, siglen;
    struct hi_pdu* parent;
  
    sublen = resp->ad.stomp.subsc ? (strchr(resp->ad.stomp.subsc, '\n') - resp->ad.stomp.subsc) : 0;
    midlen = resp->ad.stomp.msg_id ?(strchr(resp->ad.stomp.msg_id, '\n')- resp->ad.stomp.msg_id) : 0;
    siglen = resp->ad.stomp.zx_rcpt_sig ? (strchr(resp->ad.stomp.zx_rcpt_sig, '\n') - resp->ad.stomp.zx_rcpt_sig) : 0;
  
    D("NACK subsc(%.*s) msg_id(%.*s) zx_rcpt_sig(%.*s)", sublen, sublen?resp->ad.stomp.subsc:"", midlen, midlen?resp->ad.stomp.msg_id:"", siglen, siglen?resp->ad.stomp.zx_rcpt_sig:"");
    
    ASSERTOPP(resp->req, ==, 0);
    if (!stomp_find_pending_req_for_resp(io, resp)) {
      ERR("Unsolicited NACK subsc(%.*s) msg_id(%.*s)", sublen, sublen?resp->ad.stomp.subsc:"", midlen, midlen?resp->ad.stomp.msg_id:"");
      return;
    }
    parent = resp->parent;
    ASSERT(parent);
    
    /* *** add validation of zx_rcpt_sig. Lookup the cert using metadata for the EID. */
    /* Remember NACK somewhere? */
    hi_free_resp(hit, resp, "nack ");
    
    ++(parent->ad.delivb.nacks);
    if (--(parent->ad.delivb.acks) <= 0) {
      ASSERTOPI(parent->ad.delivb.acks, ==, 0);
      close_file(parent->ad.delivb.ack_fd, "got_nack");
      D("nack: freeing parent(%p)", parent);
      hi_free_req(hit, parent, "nack ");
    }
  }
  
  /*() Process ACK response from client to MESSAGE request sent by server. */
  
  /* Called by:  stomp_decode */
  static void stomp_got_ack(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* resp)
  {
    int sublen, midlen, siglen, ver;
    struct hi_pdu* parent;
    char* eid;
    char buf[1024];
    
    /* First, it was wrong for hi_add_to_reqs() to be called for an ACK as acks are
     * really responses to MESSAGEs. Thus undo that action here. */
    
    hi_del_from_reqs(io, resp);
    
    sublen = resp->ad.stomp.subsc ? (strchr(resp->ad.stomp.subsc, '\n') - resp->ad.stomp.subsc) : 0;
    midlen = resp->ad.stomp.msg_id ?(strchr(resp->ad.stomp.msg_id, '\n')- resp->ad.stomp.msg_id) : 0;
    siglen = resp->ad.stomp.zx_rcpt_sig ? (strchr(resp->ad.stomp.zx_rcpt_sig, '\n') - resp->ad.stomp.zx_rcpt_sig) : 0;
  
    DD("ACK subsc(%.*s) msg_id(%.*s) zx_rcpt_sig(%.*s)", sublen, sublen?resp->ad.stomp.subsc:"", midlen, midlen?resp->ad.stomp.msg_id:"", siglen, siglen?resp->ad.stomp.zx_rcpt_sig:"");
  
    if (!stomp_find_pending_req_for_resp(io, resp)) {
      ERR("Unsolicited ACK subsc(%.*s) msg_id(%.*s)", sublen, sublen?resp->ad.stomp.subsc:"", midlen, midlen?resp->ad.stomp.msg_id:"");
      return;
    }
    parent = resp->parent;
    ASSERT(parent);
    
    if (errmac_debug>1)
      D("ACK par_%p->len=%d rq_%p->len=%d\nparent->body(%.*s)\n   req->body(%.*s)", parent, parent->ad.delivb.len, resp->req, resp->req->ad.stomp.len, parent->ad.delivb.len, parent->ad.delivb.body, resp->req->ad.stomp.len, resp->req->ad.stomp.body);
    else
      D("ACK par_%p->len=%d rq_%p->len=%d", parent, parent->ad.delivb.len, resp->req, resp->req->ad.stomp.len);
  
    eid = zxid_my_ent_id_cstr(zx_cf);
    ver = zxbus_verify_receipt(zx_cf, io->ent->eid,
  			     siglen, siglen?resp->ad.stomp.zx_rcpt_sig:"",
  			     -2, resp->req->ad.stomp.msg_id,
  			     -2, resp->req->ad.stomp.dest,
  			     -1, eid,  /* our eid, the receipt was issued to us */
  			     resp->req->ad.stomp.len, resp->req->ad.stomp.body);
    ZX_FREE(zx_cf->ctx, eid);
    if (ver != ZXSIG_OK) {
      ERR("ACK signature validation failed: %d", ver);
      hi_free_resp(hit, resp, "ack ");
      return;
    }
    
    /* Record the receipt in /var/zxid/bus/ch/DEST/.ack/SHA1.ack for our audit trail, and to
     * indicate that we need not attempt delivery again to this entity. */
    write_all_fd_fmt(parent->ad.delivb.ack_fd, "ACK", sizeof(buf), buf, "AB1 %s ACK %.*s\n",
  		   io->ent->eid, siglen, siglen?resp->ad.stomp.zx_rcpt_sig:"");
    
    hi_free_resp(hit, resp, "ack ");
    
    if (--(parent->ad.delivb.acks) <= 0) {
      ASSERTOPI(parent->ad.delivb.acks, ==, 0);
      close_file(parent->ad.delivb.ack_fd, "got_ack");
      if (!parent->ad.delivb.nacks) {
        D("Delivered to all: mv msg to .del par_%p", parent);
        zxbus_retire(hit, parent);
      }
      hi_free_req(hit, parent, "parent ");
    }
  }
  
  /* Called by:  stomp_decode */
  static void stomp_got_subsc(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    HI_SANITY(hit->shf, hit);
    if (zxbus_subscribe(hit, io, req)) {
      /* N.B. The receipt was already sent. It needs to be sent before
       * scheduling pending deliveries, lest the simple listener clients
       * get confused by seeing a MESSAGE when expecting RECEIPT. */
    } else {
      ERR("Subscribe Problem. Disk full? %d", 0);
    }
  }
  
  /* Called by:  stomp_decode */
  static void stomp_got_unsubsc(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    stomp_cmd_ni(hit,io,req,"UNSUBSCRIBE\n");
  }
  
  /*() Nonstandard STOMP command for ZXBUS testing.
   * Based on different body content, diffrent magic can be invoked:
   * dump - dump data structures to stdout with hi_dump() */
  
  /* Called by:  stomp_decode */
  static void stomp_got_zxctl(struct hi_thr* hit, struct hi_io* io, struct hi_pdu* req)
  {
    D("ZXCTL(%.*s)", req->ad.stomp.len, req->ad.stomp.body);
    if (!memcmp(req->ad.stomp.body, "dump", sizeof("dump")-1)) hi_dump(hit->shf);
    stomp_send_receipt(hit, io, req);
  }
  
  /*() Parse STOMP 1.1 header to a struct. */
  
  /* Called by:  stomp_decode, stomp_parse_pdu */
  void stomp_parse_header(struct hi_pdu* req, char* hdr, char* val)
  {
  #define HDR(header, field, valu) } else if (!memcmp(hdr, header, sizeof(header)-1)) { if (!req->ad.stomp.field) req->ad.stomp.field = (valu)
  
    if (!memcmp(hdr, "content-length:", sizeof("content-length:")-1)) {
      if (!req->ad.stomp.len) {
        req->ad.stomp.len = atoi(val);
        req->need = req->ad.stomp.len + val - req->m + 3;  /* not accurate if more headers follow, but fix below */
      }
      DD("len=%d need=%d (%.*s)", req->ad.stomp.len, req->need, val-hdr-1, hdr);
    } else if (!memcmp(hdr, "content-type:", sizeof("content-type:")-1)) { /* ignore */
    } else if (!memcmp(hdr, "message:", sizeof("message:")-1)) { /* ignore */
    HDR("receipt:",        receipt,   val);
    HDR("destination:",    dest,      val);
    HDR("zx-rcpt-sig:",    zx_rcpt_sig, val);
    HDR("host:",           host,      val);
    HDR("version:",        vers,      val);
    HDR("accept-version:", acpt_vers, val);
    HDR("server:",         server,    val);
    HDR("heart-beat:",     heart_bt,  val);
    HDR("login:",          login,     val);
    HDR("passcode:",       pw,        val);
    HDR("session:",        session,   val);
    HDR("transaction:",    tx_id,     val);
    HDR("id:",             subs_id,   val);
    HDR("subscription:",   subsc,     val);
    HDR("ack:",            ack,       val);
    HDR("message-id:",     msg_id,    val);
    HDR("receipt-id:",     rcpt_id,   val);  DD("receipt-id(%.*s)", 4, req->ad.stomp.rcpt_id);
    } else {
      D("Unknown header(%s) ignored.", hdr);
    }
  }
  
  /*() STOMP decoder and dispatch.
   * Return:: 0 for no error (including need more and PDU complete and processed),
   * 1 to force closing connection. */
  
  /* Called by:  hi_read x2 */
  int stomp_decode(struct hi_thr* hit, struct hi_io* io)
  {
    struct hi_pdu* req = io->cur_pdu;
    char* command;
    char* hdr;
    char* val;
    char* p = req->m;
  
    D("decode req(%p)->need=%d (%.*s)", req, req->need, (int)MIN(req->ap - req->m, 3), req->m);
    
    HI_SANITY(hit->shf, hit);
    
    /* Skip newlines and check size */
  
    for (; p < req->ap && ONE_OF_2(*p, '\n', '\r'); ++p) ;
    
    if (req->ap - p < STOMP_MIN_PDU_SIZE) {   /* too little, need more */
      req->need = STOMP_MIN_PDU_SIZE;
      D("need=%d have=%d", req->need, (int)(req->ap - req->m));
      return  HI_NEED_MORE;
    }
  
    /* Extract command */
  
    command = p;
    hdr = memchr(p, '\n', req->ap - p);
    if (!hdr || ++hdr == req->ap) {
      req->need = MAX(STOMP_MIN_PDU_SIZE, req->ap - req->m + 2);
      D("need=%d have=%d", req->need, (int)(req->ap - req->m));
      return  HI_NEED_MORE;
    }
    memset(&req->ad.stomp, 0, sizeof(req->ad.stomp));
    p = hdr;
  
    /* Decode headers
     * 012345678901234567890
     * STOMP\n\n\0
     *         ^-p
     * STOMP\r\n\r\n\0
     *           ^-p
     * STOMP\nhost:foo\n\n\0
     *        ^-p        ^-pp
     * STOMP\r\nhost:foo\r\n\r\n\0
     *          ^-p          ^-pp
     * STOMP\nhost:foo\naccept-version:1.1\n\n\0
     *        ^-p       ^-pp                 ^-ppp
     * STOMP\r\nhost:foo\r\naccept-version:1.1\r\n\r\n\0
     *          ^-p         ^-pp                   ^-ppp
     */
  
    while (!ONE_OF_2(*p,'\n','\r')) {
      hdr = p;
      p = memchr(p, '\n', req->ap - p);
      if (!p || ++p == req->ap) {
        req->need = MAX(STOMP_MIN_PDU_SIZE, req->ap - req->m + 2);
        D("need=%d have=%d", req->need, (int)(req->ap - req->m));
        return HI_NEED_MORE;
      }
      val = memchr(hdr, ':', p-hdr);
      if (!val)
        return stomp_frame_err(hit, io, req, "Header missing colon.");
      ++val; /* skip : */
      stomp_parse_header(req, hdr, val);
    }
    
    /* Now body */
  
    if (*p == '\r') ++p;
    req->ad.stomp.body = ++p;
  
    /* p now points to first byte of body (after \n\n; at nul if no body) */
  
    if (req->ad.stomp.len) {
      req->need = p - req->m + req->ad.stomp.len + 1 /* nul */;  /* req->need has to be set correctly for hi_checkmore() to work right. */
      if (req->ad.stomp.len < req->ap - p) {
        /* Got complete with content-length */
        p += req->ad.stomp.len;
        if (*p++)
  	return stomp_frame_err(hit, io, req, "No nul to terminate body.");
      } else {
        D("need=%d have=%d", req->need, (int)(req->ap - req->m));
        return HI_NEED_MORE;
      }
    } else {
      /* Scan until nul */
      while (1) {
        if (req->ap - p < 1) {   /* too little, need more */
  	req->need = req->ap - req->m + 1;
  	D("need=%d have=%d", req->need, (int)(req->ap - req->m));
  	return HI_NEED_MORE;
        }
        if (!*p++) {  /* nul termination found */
  	req->ad.stomp.len = p - req->ad.stomp.body - 1;
  	req->need = p - req->m;
  	break;
        }
      }
    }
    
    HI_SANITY(hit->shf, hit);
    hi_add_to_reqs(hit, io, req, STOMP_MIN_PDU_SIZE);
    HI_SANITY(hit->shf, hit);
  
    /* Command dispatch */
  
  #define CMD(vrb, action) } else if (!memcmp(command, vrb, sizeof(vrb)-1)) { action
   
    if (!memcmp(command, "STOMP", sizeof("STOMP")-1)
        || (!memcmp(command, "CONNECT", sizeof("CONNECT")-1)
  	  &&ONE_OF_2(command[sizeof("CONNECT")],'\n','\r')))
    { return stomp_got_login(hit,io,req);
    CMD("SEND",        stomp_got_send(hit,io,req) );
    CMD("ACK",         stomp_got_ack(hit,io,req) );
    CMD("DISCONNECT",  return stomp_got_disc(hit,io,req) );
    CMD("SUBSCRIBE",   stomp_got_subsc(hit,io,req) );
    CMD("UNSUBSCRIBE", stomp_got_unsubsc(hit,io,req) );
    CMD("BEGIN",       stomp_cmd_ni(hit,io,req,p) );
    CMD("COMMIT",      stomp_cmd_ni(hit,io,req,p) );
    CMD("ABORT",       stomp_cmd_ni(hit,io,req,p) );
    CMD("NACK",        stomp_got_nack(hit,io,req) );
    /* Commands client would see (sent by server) */
    CMD("CONNECTED",   stomp_cmd_ni(hit,io,req,p) );
    CMD("MESSAGE",     stomp_cmd_ni(hit,io,req,p) );
    CMD("RECEIPT",     stomp_cmd_ni(hit,io,req,p) );
    CMD("ERROR",       return stomp_got_err(hit,io,req) );
    CMD("ZXCTL",       stomp_got_zxctl(hit,io,req) );  /* Custom command */
    } else {
      D("Unknown command(%.*s) ignored.", 4, command);
      stomp_cmd_ni(hit,io,req,p);
    }
    return 0;
  }
  
  /*() Parse PDU to extract headers. Typically this is called
   * to recover a persisted PDU, i.e. we can assume to have
   * all the data at hand already. This simplifies error reporting.
   * return:: 0 on success, 1 on error
   * see also:: stomp_decode() is usually used for decoring network traffic */
  
  /* Called by:  zxbus_sched_pending_delivery */
  int stomp_parse_pdu(struct hi_pdu* pdu)
  {
    char* hdr;
    char* val;
    char* p;
  
    /* Parse STOMP message */
    
    memset(&pdu->ad.stomp, 0, sizeof(pdu->ad.stomp));
    p = pdu->m;
    hdr = memchr(p, '\n', pdu->ap - p);
    if (!hdr || ++hdr == pdu->ap) {
      pdu->need = MAX(STOMP_MIN_PDU_SIZE, pdu->ap - pdu->m + 2);
      ERR("PDU from file is too small. need=%d have=%d", pdu->need, (int)(pdu->ap - pdu->m));
      return 1;
    }
    p = hdr;
    
    while (!ONE_OF_2(*p,'\n','\r')) {
      hdr = p;
      p = memchr(p, '\n', pdu->ap - p);
      if (!p || ++p == pdu->ap) {
        pdu->need = MAX(STOMP_MIN_PDU_SIZE, pdu->ap - pdu->m + 2);
        ERR("need=%d have=%d", pdu->need, (int)(pdu->ap - pdu->m));
        return 1;
      }
      val = memchr(hdr, ':', p-hdr);
      if (!val) {
        ERR("Malformed PDU from file. Header missing a colon. %p", pdu);
        return 1;
      }
      ++val; /* skip : */
      stomp_parse_header(pdu, hdr, val);
    }
  
    if (*p == '\r') ++p;
    pdu->ad.stomp.body = ++p;
    
    if (pdu->ad.stomp.len) {
      pdu->need = p - pdu->m + pdu->ad.stomp.len + 1 /* nul */;
      if (pdu->ad.stomp.len < pdu->ap - p) {
        /* Got complete with content-length */
        p += pdu->ad.stomp.len;
        if (*p++) {
  	ERR("Malformed PDU from file: No nul to terminate body. %x", p[-1]);
        }
      } else {
        ERR("PDU from file has specified length(%d), but not enough data. need=%d have=%d", pdu->ad.stomp.len, pdu->need, (int)(pdu->ap - pdu->m));
        return 1;
      }
      return 0;
    } else {
      /* Scan until nul */
      while (1) {
        if (pdu->ap - p < 1) {   /* too little, need more */
  	pdu->need = pdu->ap - pdu->m + 1;
  	ERR("PDU from file without length but too short to have even nul termination. need=%d have=%d", pdu->need, (int)(pdu->ap - pdu->m));
  	return 1;
        }
        if (!*p++) {  /* nul termination found */
  	pdu->ad.stomp.len = p - pdu->ad.stomp.body - 1;
  	pdu->need = p - pdu->m;
  	return 0;
        }
      }
    }
    return 0;
  }
  
  /* EOF  --  stomp.c */
  