New audio subsystem.

[jog] / txport.c

/* -*-c-*-
 *
 * $Id: txport.c,v 1.2 2002/01/30 09:27:10 mdw Exp $
 *
 * Transport switch glue
 *
 * (c) 2001 Mark Wooding
 */

/*----- Licensing notice --------------------------------------------------* 
 *
 * This file is part of Jog: Programming for a jogging machine.
 *
 * Jog is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 * 
 * Jog is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with Jog; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

/*----- Revision history --------------------------------------------------* 
 *
 * $Log: txport.c,v $
 * Revision 1.2  2002/01/30 09:27:10  mdw
 * Transport configuration overhaul.  Configuration string is now a list of
 * `;'-separated `key=value' strings, which can be handled either by the
 * core or the transport module.  The newline character(s) are a core
 * parameter now.
 *
 * Revision 1.1  2002/01/25 19:34:45  mdw
 * Initial revision
 *
 */

/*----- Header files ------------------------------------------------------*/

#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#include <ctype.h>
#include <errno.h>
#include <stdio.h>
#include <stdarg.h>

#include <sys/types.h>
#include <sys/time.h>
#include <unistd.h>
#include <pthread.h>

#include <mLib/darray.h>
#include <mLib/dstr.h>
#include <mLib/lbuf.h>
#include <mLib/sub.h>
#include <mLib/tv.h>

#include "err.h"
#include "jog.h"
#include "txport.h"

/*----- Global variables --------------------------------------------------*/

#define TX_LIST 0
#include "tx-socket.h"
#include "tx-serial-unix.h"
txport_ops *txlist = TX_LIST;

const char *txname = 0;
const char *txfile = 0;
const char *txconf = 0;

/*----- Main code ---------------------------------------------------------*/

/* --- @newline@ --- *
 *
 * Arguments:	@char *s@ = pointer to line
 *		@size_t len@ = length of line
 *		@void *txv@ = pointer to transport context
 *
 * Returns:	---
 *
 * Use:		Adds a line to the list.
 */

static void newline(char *s, size_t len, void *txv)
{
  txport *tx = txv;
  txline *l;

  if (!s)
    return;
  T( trace(T_TX, "tx: completed line: `%s'", s); )
  l = CREATE(txline);
  l->s = xmalloc(len + 1);
  memcpy(l->s, s, len + 1);
  l->len = len;
  l->tx = tx;
  l->next = 0;
  l->prev = tx->ll_tail;
  if (tx->ll_tail)
    tx->ll_tail->next = l;
  else
    tx->ll = l;
  tx->ll_tail = l;
}

/* --- @tx_configure@ --- *
 *
 * Arguments:	@txport *tx@ = pointer to transport block
 *		@const char *config@ = config string
 *
 * Returns:	Zero if OK, nonzero on errors.
 *
 * Use:		Applies a configuration string to a transport.
 */

int tx_configure(txport *tx, const char *config)
{
  char *c;
  char *k, *v;
  int rc = -1;

  if (!config)
    return (0);
  c = xstrdup(config);
  for (k = strtok(c, ";"); k; k = strtok(0, ";")) {
    if ((v = strchr(k, '=')) != 0)
      *v++ = 0;
    if (strcmp(k, "nl") == 0 || strcmp(k, "newline") == 0) {
      int d;
      if (!v)
	d = LBUF_CRLF;
      else if (strcmp(v, "none") == 0)
	d = 0;
      else if (strcmp(v, "crlf") == 0)
	d = LBUF_CRLF;
      else if (strcmp(v, "crlf-strict") == 0 ||
	       strcmp(v, "strict-crlf") == 0)
	d = LBUF_STRICTCRLF;
      else if (strcmp(v, "cr") == 0)
	d = '\r';
      else if (strcmp(v, "lf") == 0)
	d = '\n';
      else if (v[0] == '\\') switch (v[1]) {
	case 0: d = '\\'; break;
	case 'a': d = 0x07; goto d_single;
	case 'b': d = 0x08; goto d_single;
	case 'f': d = 0x0c; goto d_single;
	case 'n': d = 0x0a; goto d_single;
	case 'r': d = 0x0d; goto d_single;
	case 't': d = 0x09; goto d_single;
	case 'v': d = 0x0b; goto d_single;
	case 'e': d = 0x1b; goto d_single;
	case 'x':
	  if (!isxdigit((unsigned char)v[2]) ||
	      (d = strtoul(v + 2, &v, 16) || *v)) {
	    err_report(ERR_TXPORT, ERRTX_CONFIG, 0,
		       "bad hex escape `%s' in `newline' config", v);
	    goto err;
	  }
	  break;
	default:
	  if (isdigit((unsigned char)v[0])) {
	    d = strtoul(v + 1, &v, 8);
	    if (*v) {
	      err_report(ERR_TXPORT, ERRTX_CONFIG, 0,
			 "bad octal escape `%s' in `newline' config", v);
	      goto err;
	    }
	  }
	  d = v[1];
      d_single:
	  if (v[2]) {
	    err_report(ERR_TXPORT, ERRTX_CONFIG, 0,
		       "unknown escape `%s' in `newline' config", v);
	    goto err;
	  }
	  break;
      } else if (v[1] == 0)
	d = v[0];
      else {
	err_report(ERR_TXPORT, ERRTX_CONFIG, 0,
		   "unknown delimiter `%s' in `newline' config", v);
	goto err;
      }
      tx->lb.delim = d;
    } else {
      int e = 0;
      if (tx->ops->configure)
	e = tx->ops->configure(tx, k, v);
      if (!e) {
	err_report(ERR_TXPORT, ERRTX_CONFIG, 0,
		   "unrecognized configuration keyword `%s'", k);
      }
      if (e <= 0)
	goto err;
    }
  }

  rc = 0;
err:
  xfree(c);
  return (rc);
}

/* --- @tx_create@ --- *
 *
 * Arguments:	@const char *name@ = name of transport to instantiate
 *		@const char *file@ = filename for transport
 *		@const char *config@ = config string
 *
 * Returns:	A pointer to the transport context, or null on error.
 *
 * Use:		Creates a new transport.
 */

txport *tx_create(const char *name, const char *file, const char *config)
{
  txport_ops *o;
  txport *tx;
  pthread_attr_t ta;
  dstr d = DSTR_INIT;
  size_t len;
  int e;

  /* --- Look up the transport by name --- */

  if (!name) {
    o = txlist;
    goto found;
  }
  len = strlen(name);
  for (o = txlist; o; o = o->next) {
    if (strncmp(name, o->name, len) == 0)
      goto found;
  }
  err_report(ERR_TXPORT, ERRTX_BADTX, 0, "unknown transport `%s'", name);
  return (0);

  /* --- Set up the transport block --- */

found:
  if (!file) {
    const struct txfile *fv;
    for (fv = o->fv; fv->env || fv->name; fv++) {
      if (fv->env && (file = getenv(fv->env)) == 0)
	continue;
      DRESET(&d);
      if (file)
	DPUTS(&d, file);
      if (file && fv->name)
	DPUTC(&d, '/');
      if (fv->name)
	DPUTS(&d, fv->name);
      break;
    }
    file = d.buf;
  }
  if (!config)
    config = o->config;
  if ((tx = o->create(file)) == 0)
    goto fail_0;
  tx->ops = o;
  lbuf_init(&tx->lb, newline, tx);
  if (tx_configure(tx, config))
    goto fail_1;
  tx->ll = 0;
  tx->ll_tail = 0;
  if ((e = pthread_mutex_init(&tx->mx, 0)) != 0) {
    err_report(ERR_TXPORT, ERRTX_CREATE, e,
	       "mutex creation failed: %s", strerror(e));
    goto fail_1;
  }
  if ((e = pthread_cond_init(&tx->cv, 0)) != 0) {
    err_report(ERR_TXPORT, ERRTX_CREATE, e,
	       "condvar creation failed: %s", strerror(e));
    goto fail_2;
  }
  if ((e = pthread_attr_init(&ta)) != 0) {
    err_report(ERR_TXPORT, ERRTX_CREATE, e,
	       "thread attribute creation failed: %s", strerror(e));
    goto fail_3;
  } 
  if ((e = pthread_attr_setdetachstate(&ta, PTHREAD_CREATE_DETACHED)) ||
      (e = pthread_create(&tx->tid, &ta, tx->ops->fetch, tx)) != 0) {
    err_report(ERR_TXPORT, ERRTX_CREATE, e,
	       "thread creation failed: %s", strerror(e));
    goto fail_4;
  }
  pthread_attr_destroy(&ta);
  DA_CREATE(&tx->buf);
  tx->s = TX_READY;
  DDESTROY(&d);
  return (tx);

  /* --- Something went wrong --- */

fail_4:
  pthread_attr_destroy(&ta);
fail_3:
  pthread_cond_destroy(&tx->cv);
fail_2:
  pthread_mutex_destroy(&tx->mx);
fail_1:
  lbuf_destroy(&tx->lb);
  tx->ops->destroy(tx);
fail_0:
  DDESTROY(&d);
  return (0);
}

/* --- @tx_write@ --- *
 *
 * Arguments:	@txport *tx@ = pointer to transport context
 *		@const void *p@ = pointer to buffer to write
 *		@size_t sz@ = size of buffer
 *
 * Returns:	Zero if OK, or @-1@ on error.
 *
 * Use:		Writes some data to a transport.
 */

int tx_write(txport *tx, const void *p, size_t sz)
{
  T( trace_block(T_TX, "tx: outgoing data", p, sz); )
  if (tx->ops->write(tx, p, sz) < 0) {
    err_report(ERR_TXPORT, ERRTX_WRITE, errno,
	       "error writing to transport: %s", strerror(errno));
    return (-1);
  }
  return (0);
}

/* --- @tx_printf@ --- *
 *
 * Arguments:	@txport *tx@ = pointer to transport context
 *		@const char *p@ = pointer to string to write
 *
 * Returns:	The number of characters printed, or @EOF@ on error.
 *
 * Use:		Writes a textual message to a transport.
 */

int tx_vprintf(txport *tx, const char *p, va_list *ap)
{
  dstr d = DSTR_INIT;
  int rc;

  dstr_vputf(&d, p, *ap);
  rc = d.len;
  rc = tx_write(tx, d.buf, d.len);
  DDESTROY(&d);
  return (rc);
}

int tx_printf(txport *tx, const char *p, ...)
{
  va_list ap;
  int rc;

  va_start(ap, p);
  rc = tx_vprintf(tx, p, &ap);
  va_end(ap);
  return (rc);
}

/* --- @tx_newline@ --- *
 *
 * Arguments:	@txport *tx@ = pointer to transport context
 *
 * Returns:	Zero if OK, nonzero on error.
 *
 * Use:		Writes a newline (record boundary) to the output.
 */

int tx_newline(txport *tx)
{
  static const char crlf[2] = { 0x0d, 0x0a };
  char c;
  const char *p;
  size_t sz;

  switch (tx->lb.delim) {
    case LBUF_CRLF:
    case LBUF_STRICTCRLF:
      p = crlf;
      sz = 2;
      break;
    case 0:
      return (0);
    default:
      c = tx->lb.delim;
      p = &c;
      sz = 1;
      break;
  }
  return (tx_write(tx, p, sz));
}

/* --- @tx_read@, @tx_readx@ --- *
 *
 * Arguments:	@txport *tx@ = pointer to transport context
 *		@unsigned long t@ = time to wait for data (ms)
 *		@int (*filter)(const char *s, void *p)@ = filtering function
 *		@void *p@ = pointer argument for filter
 *
 * Returns:	A pointer to a line block, which must be freed using
 *		@tx_freeline@.
 *
 * Use:		Fetches a line from the buffer.  Each line is passed to the
 *		filter function in oldest-to-newest order; the filter
 *		function returns nonzero to choose a line.  If no suitable
 *		line is waiting in the raw buffer, the program blocks while
 *		more data is fetched, until the time limit @t@ is exceeded,
 *		in which case a null pointer is returned.  A null filter
 *		function is equivalent to one which always selects its line.
 */

txline *tx_readx(txport *tx, unsigned long t,
		 int (*filter)(const char *s, void *p), void *p)
{
  txline *l, **ll = &tx->ll;
  int e;
  struct timeval now, tv;
  struct timespec ts;
  unsigned f = 0;

#define f_lock 1u

  /* --- Get the time to wait until --- */

  T( trace(T_TXSYS, "txsys: tx_readx begin"); )
  if (t != FOREVER) {
    gettimeofday(&now, 0);
    tv_addl(&tv, &now, t / 1000, (t % 1000) * 1000);
    ts.tv_sec = tv.tv_sec;
    ts.tv_nsec = tv.tv_usec * 1000;
  }

  /* --- Check for a matching line --- */

again:
  for (; *ll; ll = &l->next) {
    l = *ll;
    if (!filter || filter(l->s, p)) {
      T( trace(T_TXSYS, "txsys: matched line; done"); )
      goto done;
    }
  }
  l = 0;

  /* --- Lock the buffer --- *
   *
   * The following operations require a lock on the buffer, so we obtain that
   * here.
   */

  if (!(f & f_lock)) {
    if ((e = pthread_mutex_lock(&tx->mx)) != 0) {
      err_report(ERR_TXPORT, ERRTX_READ, e,
		 "error locking mutex: %s", strerror(errno));
      goto done;
    }
    f |= f_lock;
    T( trace(T_TXSYS, "txsys: locked buffer"); )
  }

  /* --- Push more stuff through the line buffer --- */

check:
  if (DA_LEN(&tx->buf)) {
    T( trace_block(T_TX, "tx: incoming data",
		   DA(&tx->buf), DA_LEN(&tx->buf)); )
    if (tx->lb.delim) 
      lbuf_snarf(&tx->lb, DA(&tx->buf), DA_LEN(&tx->buf));
    else
      newline((char *)DA(&tx->buf), DA_LEN(&tx->buf), tx);
    DA_SHRINK(&tx->buf, DA_LEN(&tx->buf));
    goto again;
  }

  /* --- If nothing else can arrive, give up --- */

  if (tx->s == TX_CLOSE) {
    lbuf_close(&tx->lb);
    T( trace(T_TXSYS, "txsys: transport closed; flushing"); )
    tx->s = TX_CLOSED;
    goto again;
  }
  if (!t || tx->s == TX_CLOSED) {
    T( trace(T_TX, "tx: transport is closed"); )
    goto done;
  }

  /* --- Wait for some more data to arrive --- */

  T( trace(T_TXSYS, "txsys: waiting for data"); )
  if (t == FOREVER)
    e = pthread_cond_wait(&tx->cv, &tx->mx);
  else {
    gettimeofday(&now, 0);
    if (TV_CMP(&now, >=, &tv)) {
      T( trace(T_TXSYS, "txsys: timed out"); )
      goto done;
    }
    e = pthread_cond_timedwait(&tx->cv, &tx->mx, &ts);
  }
  if (e && e != ETIMEDOUT && e != EINTR) {
    err_report(ERR_TXPORT, ERRTX_READ, e,
	       "error waiting on condvar: %s", strerror(errno));
    goto done;
  }
  T( trace(T_TXSYS, "txsys: woken, checking again"); )
  goto check;

  /* --- Everything is finished --- */

done:
  if (f & f_lock) {
    pthread_mutex_unlock(&tx->mx);
    T( trace(T_TXSYS, "txsys: unlock buffer"); )
  }
  T( trace(T_TXSYS, "tx_readx done"); )
  return (l);

#undef f_lock
}

txline *tx_read(txport *tx, unsigned long t)
{
  return (tx_readx(tx, t, 0, 0));
}

/* --- @tx_freeline@ --- *
 *
 * Arguments:	@txline *l@ = pointer to line
 *
 * Returns:	---
 *
 * Use:		Frees a line block.
 */

void tx_freeline(txline *l)
{
  txport *tx = l->tx;
  if (l->next)
    l->next->prev = l->prev;
  else
    tx->ll_tail = l->prev;
  if (l->prev)
    l->prev->next = l->next;
  else
    tx->ll = l->next;
  xfree(l->s);
  DESTROY(l);
}

/* --- @tx_destroy@ --- *
 *
 * Arguments:	@txport *tx@ = transport context
 *
 * Returns:	---
 *
 * Use:		Destroys a transport.
 */

void tx_destroy(txport *tx)
{
  txline *l, *ll;

  if (tx->s == TX_READY) {
    pthread_mutex_lock(&tx->mx);
    if (tx->s == TX_READY)
      pthread_cancel(tx->tid);
    pthread_mutex_unlock(&tx->mx);
  }
  pthread_mutex_destroy(&tx->mx);
  pthread_cond_destroy(&tx->cv);
  DA_DESTROY(&tx->buf);
  lbuf_destroy(&tx->lb);
  for (l = tx->ll; l; l = ll) {
    ll = l->next;
    xfree(l->s);
    DESTROY(l);
  }
  tx->ops->destroy(tx);    
}

/*----- That's all, folks -------------------------------------------------*/