/*b
 * Copyright (C) 2001,2002  Rick Richardson
 *
 * This program 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.
 *
 * This program 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 this program; if not, write to the Free Software
 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 *
 * Author: Rick Richardson <rickr@mn.rr.com>
b*/

/*
 * esignal.com
 *
 * Binary, single connection streamer.  Uses linuxtrade.auth to
 * perform authorization.
 *
 * Not a bad protocol overall.  Missing some features like top 10
 * and new hi/lows.  There are some bugs, like always sending the
 * stock fullname in the quote update records (wasteful).
 *
 */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <ncurses.h>
#include <panel.h>
#include <errno.h>
#include <time.h>
#include <sys/time.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include "error.h"
#include "debug.h"
#include "p2open.h"
#include "rc.h"
#include "streamer.h"
#include "linuxtrade.h"
#include "optchain.h"
#include "info.h"
#include "util.h"
#include "chart.h"

extern int Debug;

#define		BUFLEN		65536
#define		SYMCNT		200
#define		SYMBUFLEN	(SYMCNT * (SYMLEN+1) + 1)

typedef struct streamerpriv
{
	FILE	*afp[2];
	char	hostname[128];
	char	key[128];

	char	buf[BUFLEN+1];
	char	*bufp;
	char	*bufe;

	time_t	last_keepalive;
} STREAMERDATA;

typedef struct
{
	char	*canon, *sym;
} SYMMAP;

static SYMMAP SymMap[] =
{
	{	"$DJI",		"$INDU",	},
	{	"$DJT",		"$TRAN",	},
	{	"$DJU",		"$UTIL",	},

	{	"$NYA",		"$NYA",		},
	{	"$TRIN",	"$TRIN",	},
	{	"$TICK",	"$TICK",	},

	{	"$COMP",	"$COMPQ",	},
	{	"$NDX",		"$NDX",		},
	{	"$SPX",		"$SPX",		},
	{	"$OEX",		"$OEX",		},
	{	"$MID",		"$MID",		},
	{	"$SML",		"$SML",		},
	{	"$RLX",		"$RLX",		},

	{	"$XAL",		"$XAL",		},
	{	"$BTK",		"$BTK",		},
	{	"$XBD",		"$XBD",		},
	{	"$XAX",		"$XAX",		},
	{	"$XCI",		"$XCI",		},
	{	"$IIX",		"$IIX",		},
	{	"$NWX",		"$NWX",		},
	{	"$XOI",		"$XOI",		},
	{	"$DRG",		"$DRG",		},
	{	"$XTC",		"$XTC",		},

	{	"$GSO",		"$GSO",		},
	{	"$HWI",		"$HWI",		},
	{	"$RUI",		"$RUI",		},
	{	"$RUT",		"$RUT",		},
	{	"$RUA",		"$RUA",		},
	{	"$SOX",		"$SOX",		},
	{	"$OSX",		"$OSX",		},

	{	"$BKX",		"$BKX",		},
	{	"$GOX",		"$GOX",		},
	{	"$XAU",		"$XAU",		},
	{	"$YLS",		"$YLS",		},
	{	"$DDX",		"$DDX",		},
	{	"$DOT",		"$DOT",		},

	{	"$RXP",		"$RXP",		},
	{	"$RXH",		"$RXH",		},
	{	"$XNG",		"$XNG",		},
	{	"$FPP",		"$FPP",		},
	{	"$DJR",		"$DJR",		},
	{	"$UTY",		"$UTY",		},

	{	"$TYX",		"$TYX",		},
	{	"$TRINQ",	"$TRINQ",	},
	{	"$TICKQ",	"$TICKQ",	},
	{	"$TRINA",	"$TRINA",	},
	{	"$TICKA",	"$TICKA",	},
	{	"$VIX",		"$VIX",		},
	{	"$VXN",		"$VXN",		},
	{	NULL,		NULL		}
};

//
// Convert canonical index names to/from streamer index names
//
static void
streamer_canon2sym(char *out, char *in)
{
	char	*ip, *op;
	char	*p;
	int	len;
	SYMMAP	*map;
	char	sep = '|';

	ip = in;
	op = out;
	for (;;)
	{
		p = strchr(ip, '|');
		if (!p) p = strchr(ip, ',');
		if (!p) p = strchr(ip, ' ');
		if (!p) p = strchr(ip, 0);

		len = p - ip;
		if (ip[len-1] != '.')
		{
			memcpy(op, ip, len);
			op[len] = 0;
		}
		else
		{
			// option symbol, insert blank after root
			sprintf(op, "%.*s %.2s", len-2-1, ip, ip+len-2-1);
		}

		for (map = SymMap; map->canon; ++map)
			if (strcmp(op, map->canon) == 0)
			{
				strcpy(op, map->sym);
				break;
			}

		if (*p == 0)
			break;

		ip += len + 1;
		op = strchr(op, 0);
		*op++ = *p ? sep : 0;
		*op = 0;
	}
}

static void
streamer_sym2canon(char *out, char *in)
{
	SYMMAP	*map;
	char	*p;

	if (strchr(in, ' '))
	{
		// Implies option symbol
		char buf[SYMLEN+1];
		strcpy(buf, in);
		in = buf;

		for (p = in; *p; ++p)
			if (*p != ' ')
				*out++ = *p;
		*out++ = '.';
		*out = 0;
		return;
	}
	for (map = SymMap; map->sym; ++map)
		if (strcmp(in, map->sym) == 0)
		{
			strcpy(out, map->canon);
			return;
		}

	if (in != out)
		strcpy(out, in);
}

/*
 */
static int
streamer_send(STREAMER sr, char *buf, int len)
{
    if (Debug >= 5)
    {
	    timestamp(stderr);
	    hexdump(stderr, ">", "            ", buf, len);
    }
    if (StreamerLog)
	    hexdump(StreamerLog, ">", " ", buf, len);

    return write(sr->fd[0], buf, len);
}

static int
streamer_send1chart(STREAMER sr, char code, char *str1, short duration)
{
    char	buf[256];
    int		slen1 = strlen(str1);
    int		len;

    if (slen1 > (256-8))
	return -1;

    len = 1 + 1 + slen1 + 2;
    buf[0] = len >> 24;
    buf[1] = len >> 16;
    buf[2] = len >> 8;
    buf[3] = len >> 0;

    buf[4] = code;
    buf[5] = slen1;
    memcpy(buf+6, str1, slen1);
    buf[6+slen1+0] = duration >> 8;
    buf[6+slen1+1] = duration;

    return streamer_send(sr, buf, len+4);
}

static int
streamer_send2str(STREAMER sr, char code, char *str1, char *str2)
{
    char	buf[256];
    int		slen1 = strlen(str1);
    int		slen2 = strlen(str2);
    int		len;

    if ((slen1+slen2) > (256-6))
	return -1;

    len = 1 + 1 + slen1 + 1 + slen2;
    buf[0] = len >> 24;
    buf[1] = len >> 16;
    buf[2] = len >> 8;
    buf[3] = len >> 0;

    buf[4] = code;
    buf[5] = slen1;
    memcpy(buf+6, str1, slen1);
    buf[6+slen1] = slen2;
    memcpy(buf+6+slen1+1, str2, slen2);

    return streamer_send(sr, buf, len+4);
}

static int
streamer_send1sym(STREAMER sr, char code, char *str1)
{
    char	buf[256];
    int		slen1 = strlen(str1);
    int		len;

    if (slen1 > (256-7))
	return -1;

    len = 1 + 1 + 1 + slen1;
    buf[0] = len >> 24;
    buf[1] = len >> 16;
    buf[2] = len >> 8;
    buf[3] = len >> 0;

    buf[4] = code;
    buf[5] = 0x01;
    buf[6] = slen1;
    memcpy(buf+7, str1, slen1);

    return streamer_send(sr, buf, len+4);
}

static int
streamer_select(STREAMER sr,
		int numfd, fd_set *readfds, fd_set *writefds,
		fd_set *exceptfds, struct timeval *timeout)
{
	if (readfds && FD_ISSET(sr->fd[0], readfds) && sr->priv->bufp)
	{
			FD_ZERO(readfds);
			FD_SET(sr->fd[0], readfds);
			if (writefds)
				FD_ZERO(writefds);
			if (exceptfds)
				FD_ZERO(exceptfds);
			return 1;
	}

	if (0)
		fprintf(stderr, "do select\n");
	return select(numfd, readfds, writefds, exceptfds, timeout);
}

static void
streamer_close(STREAMER sr)
{
	close(sr->fd[0]);
	sr->fd[0] = -1;
}

static void
streamer_record(STREAMER sr, FILE *fp)
{
	sr->writefile = fp;
}

static void
streamer_init(STREAMER sr)
{
	sr->refresh = 60;
	sr->fd[0] = -1;
	sr->nfd = 1;
	sr->id[0] = 0;

	sr->priv->bufp = NULL;
	sr->priv->bufe = NULL;
	time(&sr->priv->last_keepalive);
}

/*
 * Open connection to the streamer
 */
static int
streamer_open(STREAMER sr, RCFILE *rcp, FILE *readfile)
{
	struct hostent		*hep;
	struct sockaddr_in	sockaddr;
	int			rc;
	char			*p;
	unsigned char		buf[512];
	int			conn_tries;
	int			read_tries;
	char			*username = get_rc_value(rcp, "username");
	char			*password = get_rc_value(rcp, "password");
	int			port = atoi(get_rc_value(rcp, "port"));
	int			cnt;

	streamer_init(sr);
	++sr->cnt_opens;
	++sr->cnt_realopens;
	time(&sr->time_open);
	time(&sr->time_realopen);

	sr->readfile = readfile;
	if (readfile)
	{
		sr->fd[0] = fileno(readfile);
		return sr->fd[0];
	}

	/*
	 * Authenticate and get a server to connect to
	 */
	debug(5, "Authenticate...\n");
	rc = p2open("/bin/bash", PROGNAMESTR ".auth esignal.com 2>/dev/null",
				    sr->priv->afp);
	if (rc < 0)
		return SR_AUTH;

	fprintf(sr->priv->afp[1], "%s\n", username);
	fprintf(sr->priv->afp[1], "%s\n", password);
	fflush(sr->priv->afp[1]);

	fgets(sr->priv->key, sizeof(sr->priv->key), sr->priv->afp[0]);
	p = strchr(sr->priv->key, '\n'); if (p) *p = 0;
	fgets(sr->priv->hostname, sizeof(sr->priv->hostname), sr->priv->afp[0]);
	p = strchr(sr->priv->hostname, '\n'); if (p) *p = 0;

	rc = p2close(sr->priv->afp);
	if (rc < 0)
		return SR_AUTH;

	if (strncmp(sr->priv->key, "invalid", 7) == 0)
		return SR_AUTH;
	if (strncmp(sr->priv->hostname, "invalid", 7) == 0)
		return SR_AUTH;

	sprintf(sr->id, "%-.32s", sr->priv->hostname);
	p = strstr(sr->id, ".com"); if (p) *p = 0;

	/*
	 * Open connection to the streamer
	 */
	hep = mygethostbyname(sr->priv->hostname);
	if (!hep)
		return (-1);

	memcpy(&sockaddr.sin_addr, hep->h_addr, hep->h_length);
	sockaddr.sin_family = AF_INET;
	sockaddr.sin_port = htons(port);

	conn_tries = 0;
reconnect:
	if (++conn_tries >= 5)
		return -4;

	debug(5, "Open socket...\n");

	sr->fd[0] = socket(AF_INET, SOCK_STREAM, 0);
	if (sr->fd[0] < 0)
		return -2;

	debug(5, "Socket fd=%d...\n", sr->fd[0]);

	debug(5, "Connect to '%s'...\n", sr->priv->hostname);

	rc = connect_timeout(sr->fd[0], (SA *) &sockaddr, sizeof(sockaddr), 15);
	if (rc < 0)
	{
		if (Debug >= 5)
			syserror(0, "Couldn't connect\n");
		streamer_close(sr);
		sleep(1);
		goto reconnect;
	}

	sr->priv->bufp = NULL;

	debug(5, "Authorize...\n");

	rc = streamer_send2str(sr, 0x05, username, sr->priv->key);
	if (rc < 0)
	{
		streamer_close(sr);
		sleep(1);
		goto reconnect;
	}

	/* get response */
	debug(5, "Get handshake...\n");

	read_tries = 0;
	cnt = 0;
	for (;;)
	{
		fd_set		fds;
		struct timeval	tv;
		int		len;

		tv.tv_sec = 3;
		tv.tv_usec = 0;
		FD_ZERO(&fds);
		FD_SET(sr->fd[0], &fds);
		rc = streamer_select(sr, sr->fd[0]+1, &fds, NULL, NULL, &tv);
		if (rc == -1 && errno != EINTR)
			syserror(1, "Select error\n");
		if (rc == 0)
		{
			debug(5, "Timeout...\n");
			if (++read_tries == 3)
			{
				streamer_close(sr);
				sleep(2);
				goto reconnect;
			}

			continue;
		}
		if (rc <= 0)
		{
			streamer_close(sr);
			sleep(2);
			goto reconnect;
		}

		rc = streamer_read(sr, 0, buf, 4);
		if (rc != 4)
		{
		err:
			if (Debug >= 1)
				syserror(0, "Can't read streamer data\n");
			streamer_close(sr);
			sleep(2);
			goto reconnect;
		}

		len = (buf[0]<<24) + (buf[1]<<16) + (buf[2]<<8) + buf[3];
		if (len > sizeof(buf))
		{
			debug(5, "Illegal response length %d...\n", len);
			streamer_close(sr);
			sleep(2);
			goto reconnect;
		}

		rc = streamer_read(sr, 0, buf, len);
		if (rc != len)
		    goto err;

		if (++cnt == 4)
		    break;
	}

	return 0;
}

static void
streamer_send_quickquote(STREAMER sr, char *sym)
{
	char	sbuf[32];

	if (sr->fd[0] < 0 || sr->readfile)
		return;

	if (find_stock(sym))
	    return;

	streamer_canon2sym(sbuf, sym);

	streamer_send1sym(sr, 0x10, sbuf);
}

static void
streamer_send_livequote(STREAMER sr, char *sym)
{
	streamer_send_quickquote(sr, sym);
}

static void
streamer_send_symbols(STREAMER sr, char *symbols, int add)
{
	char	sbuf[SYMBUFLEN];
	char	*s, *p;

	if (sr->fd[0] < 0 || sr->readfile)
		return;

	streamer_canon2sym(sbuf, symbols);

	for (p = s = sbuf; *p; s = p+1)
	{
		int	len;
		char	buf[32];
		p = strchr(s, '|');
		if (!p) p = strchr(s, 0);
		len = p - s;
		strncpy(buf, s, len);
		buf[len] = 0;
		streamer_send1sym(sr, add ? 0x10 : 0x11, buf);
	}
}

static void
streamer_send_symbols_end(STREAMER sr, int add, int all)
{
}

static void
streamer_send_keepalive(STREAMER sr)
{
	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_printf(sr->fd[0], "_X ");
}

static void
streamer_send_disconnect(STREAMER sr)
{
	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_printf(sr->fd[0], "_1 ");
}

static void
streamer_send_top10(STREAMER sr, char market, int type, int add)
{
	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_printf(sr->fd[0], "_T,%d,%c,%d, ",
			add ? 1 : 2,
			market,
			type
			);
}

static void
streamer_send_movers(STREAMER sr, int on)
{
	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_printf(sr->fd[0], "_H%d ", on ? 1 : 2);
}

static void
streamer_send_info(STREAMER sr, char *sym, int type)
{
	char	sbuf[32];

	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_canon2sym(sbuf, sym);

	streamer_printf(sr->fd[0], "_G,%d,%s, ", type, sbuf);
}

static void
streamer_send_optchain(STREAMER sr, char *sym)
{
	if (sr->fd[0] < 0 || sr->readfile)
		return;
	return;

	streamer_printf(sr->fd[0], "_K%s ", sym);
}

static int
streamer_send_chart(STREAMER sr, char *sym, int freq, int periods, int days)
{
	char	sbuf[32];
	short	interval;

	if (sr->fd[0] < 0 || sr->readfile)
		return SR_ERR;

	streamer_canon2sym(sbuf, sym);

	switch (freq)
	{
	case FREQ_1MIN:	interval = 1; break;
	case FREQ_5MIN:	interval = 5; break;
	case FREQ_10MIN:interval = 10; break;
	case FREQ_HOUR:	interval = 60; break;
	case FREQ_DAY:	interval = -1; break;
	case FREQ_WEEK:	interval = -2; break;
	default: return -1;
	}
	streamer_send1chart(sr, 0x20, sbuf, interval);
	return 0;
}

static int
getn(unsigned char **pp, int len)
{
    unsigned char	*p = *pp;
    int			val = 0;

    while (len--)
    {
	val <<= 8;
	val += *p++;
    }
    *pp = p;
    if (val == 0x7fffFFFF)
	return 0;
    return val;
}

static double
code2scale(char *sym, int code)
{
	switch (code & 0xff)
	{
	case 'm':	return 100;
	case 'n':	return 1000;
	case 'o':	return 10000;
	case 'p':	return 100000;
	default:	
			debug(5, "%s: Unknown scale 0x%04x\n", sym, code);
			return 0;
	}
}

static void
do_fullquote(STREAMER sr, char *buf, int len)
{
	QUICKQUOTE	qq;
	LIVEQUOTE	lq;
	QUOTE		q;
	int		symlen;
	char		sym[32];
	unsigned char	*p, *e;
	int		slen;
	time_t		unixtime;
	struct tm	*tmp;
	double		scale;
	int		code;

	memset(&q, 0, sizeof(q));
	memset(&qq, 0, sizeof(qq));
	memset(&lq, 0, sizeof(lq));

	symlen = buf[1];
	p = buf + 2 + symlen;
	e = buf + len;
	if (symlen >= sizeof(sym)) symlen = sizeof(sym) - 1;
	memcpy(sym, buf+2, symlen);
	sym[symlen] = 0;
	streamer_sym2canon(q.sym, sym);
	strcpy(qq.sym, q.sym);
	strcpy(lq.sym, q.sym);

	/* or is it a string?? */ code = getn(&p, 2);
	    if (p >= e) goto out;

	scale = code2scale(q.sym, code);
	if (!scale) return;
	q.scale = scale;

	/*open*/ getn(&p, 4);
	    if (p >= e) goto out;
	qq.high = q.high = getn(&p, 4) / scale;
	    if (p >= e) goto out;

	qq.low = q.low = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	lq.last = qq.last = q.last = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	lq.close = qq.prev_close = q.close = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	qq.volume = q.volume = getn(&p, 4);
	    if (p >= e) goto out;

	q.last_size = getn(&p, 4);
	    if (p >= e) goto out;
	qq.bid = q.bid = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	qq.bid_size = q.bid_size = getn(&p, 4);
	    if (p >= e) goto out;
	qq.ask = q.ask = getn(&p, 4) / scale;
	    if (p >= e) goto out;

	qq.ask_size = q.ask_size = getn(&p, 4);
	    if (p >= e) goto out;
	/*list exch*/ slen = *p++; p += slen;
	    if (p >= e) goto out;
	/*trade exch*/ slen = *p++; p += slen;
	    if (p >= e) goto out;
	/*bid exch*/ slen = *p++; p += slen;
	    if (p >= e) goto out;

	/*ask exch*/ slen = *p++; p += slen;
	    if (p >= e) goto out;
	qq.low52 = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	qq.high52 = getn(&p, 4) / scale;
	    if (p >= e) goto out;
	unixtime = getn(&p, 4);
	tmp = localtime(&unixtime);
	qq.timetrade = q.time =
	    tmp->tm_hour*3600 + tmp->tm_min*60 + tmp->tm_sec;
	    if (p >= e) goto out;

	/* always 00 00? */ getn(&p, 2);
	    if (p >= e) goto out;
	/*fullname*/ slen = *p++;
	    memcpy(qq.fullname, p, slen); qq.fullname[slen] = 0;
	    p += slen;
	    if (p >= e) goto out;

	/* Mutual fund NAV? */ getn(&p, 4);
	    if (p >= e) goto out;
	/* always 7f000000? */ getn(&p, 4);
	    if (p >= e) goto out;
	/*tick*/ slen = *p++; p += slen;
	    if (p >= e) goto out;

out:
	display_quote(&q, 0);
	optchain_quote(&q);
	info_quickquote(&qq);
	display_livequote(&lq);

	// Turn off streaming for any livequote stocks
	if (!find_stock(q.sym))
	    streamer_send1sym(sr, 0x11, q.sym);
}

static void
do_quoteupdate(STREAMER sr, char *buf, int len)
{
	STOCK		*sp;
	QUOTE		q;
	int		symlen;
	char		sym[32];
	unsigned char	*p, *e;
	int		nfields;
	int		slen;
	time_t		unixtime;
	struct tm	*tmp;
	double		scale;
	int		code, val;
	int		tradetype = 0;

	symlen = buf[1];
	p = buf + 2 + symlen;
	e = buf + len;
	if (symlen >= sizeof(sym)) symlen = sizeof(sym) - 1;
	memcpy(sym, buf+2, symlen);
	sym[symlen] = 0;
	streamer_sym2canon(q.sym, sym);

	sp = find_stock(q.sym);
	if (!sp)
	    return;

	q = sp->cur;
	scale = q.scale;
	if (scale <= 0)
	    scale = 100000;
	nfields = getn(&p, 1);
	while (nfields-- && p < e)
	{
	    code = getn(&p, 1);
	    switch (code)
	    {
	    case 0x01: q.high = getn(&p, 4) / scale; break;
	    case 0x02: q.low = getn(&p, 4) / scale; break;
	    case 0x03: q.last = getn(&p, 4) / scale; break;
	    case 0x04: q.close = getn(&p, 4) / scale; break;
	    case 0x05:
		val = getn(&p, 4);
		if (val > q.volume)
		    tradetype = 1;
		q.volume = val;
		break;
	    case 0x06: q.last_size = getn(&p, 4); break;
	    case 0x07: q.bid = getn(&p, 4) / scale; break;
	    case 0x08: q.bid_size = getn(&p, 4); break;
	    case 0x09: q.ask = getn(&p, 4) / scale; break;
	    case 0x0a: q.ask_size = getn(&p, 4); break;
	    case 0x0b:
		// list exchange
		slen = *p++; p += slen;
		break;
	    case 0x0c:
		// trade exchange
		slen = *p++; p += slen;
		break;
	    case 0x0d:
		// bid exchange
		slen = *p++; p += slen;
		break;
	    case 0x0e:
		// ask exchange
		slen = *p++; p += slen;
		break;
	    case 0x10: /* low52 */ getn(&p, 4); break;
	    case 0x11: /* high52 */ getn(&p, 4); break;
	    case 0x12:
		unixtime = getn(&p, 4);
		tmp = localtime(&unixtime);
		q.time = tmp->tm_hour*3600 + tmp->tm_min*60 + tmp->tm_sec;
		break;
	    case 0x18:
		// fullname
		slen = *p++; p += slen;
		break;
	    case 0x19:
		// Mutual fund NAV?
		val = getn(&p, 4);
		break;
	    case 0x1a:
		// FormT trade?
		val = getn(&p, 4);
		q.last = val / scale;
		break;
	    case 0x1b:
		// bid tick
		slen = *p++;
		q.bid_tick = *p;
		p += slen;
		break;
	    case 0x1c:
		/* or is it a string?? */
		code = getn(&p, 2);
		scale = code2scale(q.sym, code);
		if (!scale) return;
		q.scale = scale;
		break;
	    default:
		debug(5, "Unknown update code %x\n", code);
		goto out;
	    }
	}
out:
	display_quote(&q, tradetype);
}

static double
getdouble(unsigned char **pp)
{
	double		d;
	unsigned char	*p = *pp;
	unsigned char	*dp = (unsigned char *) &d;
	int		endian = 0x01;
	char		*endp = (char *) &endian;

	if (endp[0] == 0x01)
	{
	    dp[7] = *p++;
	    dp[6] = *p++;
	    dp[5] = *p++;
	    dp[4] = *p++;
	    dp[3] = *p++;
	    dp[2] = *p++;
	    dp[1] = *p++;
	    dp[0] = *p++;
	}
	else
	{
	    memcpy(dp, p, 8);
	    p += 8;
	}
	*pp = p;
	return d;
}

static void
do_chart(STREAMER sr, char *buf, int len)
{
	char		sym[32];
	int		symlen;
	short		interval;
	short		nperiods;
	unsigned char	*p, *e;
	time_t		tim;
	OHLCV		rec;
	struct tm	*tmp;

	// code, slen, symbol, interval, nperiods
	// then 40 byte records (unixtime, open, hi, lo, close, vol)
	symlen = buf[1];
	p = buf + 2 + symlen;
	e = buf + len;
	if (symlen >= sizeof(sym)) symlen = sizeof(sym) - 1;
	memcpy(sym, buf+2, symlen);
	sym[symlen] = 0;
	interval = (p[0]<<8) | p[1];
	nperiods = (p[2]<<8) | p[3];
	p += 4;
	while (nperiods--)
	{
	    tim = getn(&p, 4);
	    rec.open = getdouble(&p);
	    rec.high = getdouble(&p);
	    rec.low = getdouble(&p);
	    rec.close = getdouble(&p);
	    rec.volume = getn(&p, 4);
	    tmp = localtime(&tim);
	    chart_data(sym, tmp->tm_mon+1, tmp->tm_mday, tmp->tm_year+1900,
		    tmp->tm_hour*60 + tmp->tm_min, &rec);
	    if (p >= e)
		break;
	}
	chart_data_complete(FALSE);
}

static int
streamer_process(STREAMER sr, int fdindex)
{
	int	len;
	int	rc;
	char	buf[65536];

	rc = streamer_mustread(sr, 0, buf, 4);
	if (rc != 4)
	{
	err:
		if (sr->readfile)
		{
			if (sr->fd[0] >= 0)
			{
				fclose(sr->readfile);
				sr->fd[0] = -1;
			}
			return (0);
		}
		return (-1);
	}
	if (sr->writefile)
	    fwrite(buf, 1, 4, sr->writefile);

	len = (buf[0]<<24) + (buf[1]<<16) + (buf[2]<<8) + buf[3];
	if (len > sizeof(buf))
	    goto err;

	rc = streamer_mustread(sr, 0, buf+4, len);
	if (rc != len)
	    goto err;

	if (sr->writefile)
	    fwrite(buf, 1, len, sr->writefile);

	switch (buf[4])
	{
	case 6:
		// keepalive
		streamer_send(sr, buf, len+4);
		break;
	case 0x10:
		// fullquote
		do_fullquote(sr, buf+4, len);
		break;
	case 0x12:
		// quote update
		do_quoteupdate(sr, buf+4, len);
		break;
	case 0x21:
		do_chart(sr, buf+4, len);
		break;
	}

	return (0);
}

static void
streamer_timetick(STREAMER sr, time_t now)
{
	if (now > sr->priv->last_keepalive + 5)
	{
		streamer_send_keepalive(sr);
		sr->priv->last_keepalive = now;

		if (sr->writefile)
			fflush(sr->writefile);
	}
}

static void
nullsr(STREAMER sr)
{
}

STREAMER
esignal_new(void)
{
	STREAMER	sr;

	sr = (STREAMER) malloc(sizeof(*sr));
	if (!sr)
		return NULL;
	memset(sr, 0, sizeof(*sr));

	sr->open = streamer_open;
	sr->select = streamer_select;
	sr->close = streamer_close;
	sr->record = streamer_record;
	sr->timetick = streamer_timetick;

	sr->send_quickquote = streamer_send_quickquote;
	sr->send_livequote = streamer_send_livequote;
	sr->send_livequote_end = nullsr;
	sr->send_symbols = streamer_send_symbols;
	sr->send_symbols_end = streamer_send_symbols_end;

	sr->send_disconnect = streamer_send_disconnect;
	sr->send_top10 = streamer_send_top10;
	sr->send_movers = streamer_send_movers;
	sr->send_info = streamer_send_info;
	sr->send_optchain = streamer_send_optchain;
	sr->send_chart = streamer_send_chart;

	sr->process = streamer_process;

	sr->priv = (STREAMERDATA *) malloc(sizeof(*sr->priv));
	if (!sr->priv)
	{
		free(sr);
		return NULL;
	}

	time(&sr->time_start);

	streamer_init(sr);

	return (sr);
}