mrtk_stream.c

Go to the documentation of this file.

/*------------------------------------------------------------------------------
 * mrtk_stream.c : stream I/O functions
 *
 * Copyright (C) 2026 H.SHIONO (MRTKLIB Project)
 * Copyright (C) 2023-2025 Cabinet Office, Japan
 * Copyright (C) 2024-2025 Lighthouse Technology & Consulting Co. Ltd.
 * Copyright (C) 2023-2025 Japan Aerospace Exploration Agency
 * Copyright (C) 2023-2025 TOSHIBA ELECTRONIC TECHNOLOGIES CORPORATION
 * Copyright (C) 2015- Mitsubishi Electric Corp.
 * Copyright (C) 2014 Geospatial Information Authority of Japan
 * Copyright (C) 2014 T.SUZUKI
 * Copyright (C) 2007-2023 T.TAKASU
 *
 * SPDX-License-Identifier: BSD-2-Clause
 *----------------------------------------------------------------------------*/
#include "mrtklib/mrtk_stream.h"
 
#include <ctype.h>
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
#include "mrtklib/mrtk_rcvraw.h"
#include "mrtklib/mrtk_sol.h"
#include "mrtklib/mrtk_sys.h"
#include "mrtklib/mrtk_time.h"
#include "mrtklib/mrtk_version.h"
 
/* Forward declarations for functions in rtklib (resolved at link time) */
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <unistd.h>
#define __USE_MISC
#ifndef CRTSCTS
#define CRTSCTS 020000000000
#endif
#include <arpa/inet.h>
#include <errno.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <sys/socket.h>
#include <termios.h>
 
#include "mrtklib/mrtk_trace.h"
#include "ntrip_chunk.h"
 
/* constants -----------------------------------------------------------------*/
 
#define TINTACT 200              /* period for stream active (ms) */
#define SERIBUFFSIZE 4096        /* serial buffer size (bytes) */
#define TIMETAGH_LEN 64          /* time tag file header length */
#define MAXCLI 32                /* max client connection for tcp svr */
#define MAXSTATMSG 32            /* max length of status message */
#define DEFAULT_MEMBUF_SIZE 4096 /* default memory buffer size (bytes) */
 
#define NTRIP_AGENT MRTKLIB_SOFTNAME "/" MRTKLIB_VERSION_STRING
#define NTRIP_CLI_PORT 2101                       /* default ntrip-client connection port */
#define NTRIP_SVR_PORT 80                         /* default ntrip-server connection port */
#define NTRIP_MAXRSP 32768                        /* max size of ntrip response */
#define NTRIP_MAXSTR 256                          /* max length of mountpoint string */
#define NTRIP_RSP_OK_CLI "ICY 200 OK\r\n"         /* ntrip response: client */
#define NTRIP_RSP_OK_SVR "OK\r\n"                 /* ntrip response: server */
#define NTRIP_RSP_SRCTBL "SOURCETABLE 200 OK\r\n" /* ntrip response: source table */
#define NTRIP_RSP_TBLEND "ENDSOURCETABLE"
#define NTRIP_RSP_HTTP "HTTP/"  /* ntrip response: http */
#define NTRIP_RSP_ERROR "ERROR" /* ntrip response: error */
#define NTRIP_RSP_UNAUTH "HTTP/1.0 401 Unauthorized\r\n"
#define NTRIP_RSP_ERR_PWD "ERROR - Bad Password\r\n"
#define NTRIP_RSP_ERR_MNTP "ERROR - Bad Mountpoint\r\n"
 
#define FTP_CMD "wget" /* ftp/http command */
#define FTP_TIMEOUT 30 /* ftp/http timeout (s) */
 
#define MIN(x, y) ((x) < (y) ? (x) : (y))
 
/* stream format strings (moved from rtkcmn.c) */
const char* formatstrs[32] = {                  /* stream format strings */
                              "RTCM 2",         /*  0 */
                              "RTCM 3",         /*  1 */
                              "NovAtel OEM7",   /*  2 */
                              "NovAtel OEM3",   /*  3 */
                              "u-blox UBX",     /*  4 */
                              "Superstar II",   /*  5 */
                              "Hemisphere",     /*  6 */
                              "SkyTraq",        /*  7 */
                              "Javad GREIS",    /*  8 */
                              "NVS BINR",       /*  9 */
                              "BINEX",          /* 10 */
                              "Trimble RT17",   /* 11 */
                              "Septentrio SBF", /* 12 */
                              "RINEX",          /* 13 */
                              "SP3",            /* 14 */
                              "RINEX CLK",      /* 15 */
                              "SBAS",           /* 16 */
                              "NMEA 0183",      /* 17 */
                              NULL};
 
/* macros --------------------------------------------------------------------*/
 
#define dev_t int
#define socket_t int
#define closesocket close
 
/* type definition -----------------------------------------------------------*/
 
typedef struct {               /* file control type */
    FILE* fp;                  /* file pointer */
    FILE* fp_tag;              /* file pointer of tag file */
    FILE* fp_tmp;              /* temporary file pointer for swap */
    FILE* fp_tag_tmp;          /* temporary file pointer of tag file for swap */
    char path[MAXSTRPATH];     /* file path */
    char openpath[MAXSTRPATH]; /* open file path */
    int mode;                  /* file mode */
    int timetag;               /* time tag flag (0:off,1:on) */
    int repmode;               /* replay mode (0:master,1:slave) */
    int offset;                /* time offset (ms) for slave */
    int size_fpos;             /* file position size (bytes) */
    gtime_t time;              /* start time */
    gtime_t wtime;             /* write time */
    uint32_t tick;             /* start tick */
    uint32_t tick_f;           /* start tick in file */
    long fpos_n;               /* next file position */
    uint32_t tick_n;           /* next tick */
    double start;              /* start offset (s) */
    double speed;              /* replay speed (time factor) */
    double swapintv;           /* swap interval (hr) (0: no swap) */
    rtk_lock_t lock;           /* lock flag */
} file_t;
 
typedef struct {             /* tcp control type */
    int state;               /* state (0:close,1:wait,2:connect) */
    char saddr[256];         /* address string */
    int port;                /* port */
    struct sockaddr_in addr; /* address resolved */
    socket_t sock;           /* socket descriptor */
    int tcon;                /* reconnect time (ms) (-1:never,0:now) */
    uint32_t tact;           /* data active tick */
    uint32_t tdis;           /* disconnect tick */
} tcp_t;
 
typedef struct tcpsvr_tag { /* tcp server type */
    tcp_t svr;              /* tcp server control */
    tcp_t cli[MAXCLI];      /* tcp client controls */
} tcpsvr_t;
 
typedef struct { /* tcp cilent type */
    tcp_t svr;   /* tcp server control */
    int toinact; /* inactive timeout (ms) (0:no timeout) */
    int tirecon; /* reconnect interval (ms) (0:no reconnect) */
} tcpcli_t;
 
typedef struct {      /* serial control type */
    dev_t dev;        /* serial device */
    int error;        /* error state */
    tcpsvr_t* tcpsvr; /* tcp server for received stream */
} serial_t;
 
typedef struct {                /* ntrip control type */
    int state;                  /* state (0:close,1:wait,2:connect) */
    int type;                   /* type (0:server,1:client) */
    int ver;                    /* requested version (NTRIP_VER_AUTO/1/2) */
    int ver_neg;                /* negotiated version after handshake */
    int chunked;                /* chunked transfer encoding flag */
    int v2_tried;               /* v2 attempt flag for AUTO fallback */
    int nb;                     /* response buffer size */
    char url[MAXSTRPATH];       /* url for proxy */
    char host[256];             /* host header value (addr:port) */
    char mntpnt[256];           /* mountpoint */
    char user[256];             /* user */
    char passwd[256];           /* password */
    char str[NTRIP_MAXSTR];     /* mountpoint string for server */
    uint8_t buff[NTRIP_MAXRSP]; /* response buffer */
    tcpcli_t* tcp;              /* tcp client */
    chunk_dec_t cdec;           /* chunked transfer decoder */
} ntrip_t;
 
typedef struct {                /* ntrip client/server connection type */
    int state;                  /* state (0:close,1:connect) */
    int ver;                    /* detected NTRIP version */
    char mntpnt[256];           /* mountpoint */
    char str[NTRIP_MAXSTR];     /* mountpoint string for server */
    int nb;                     /* request buffer size */
    uint8_t buff[NTRIP_MAXRSP]; /* request buffer */
} ntripc_con_t;
 
typedef struct {               /* ntrip caster control type */
    int state;                 /* state (0:close,1:wait,2:connect) */
    int type;                  /* type (0:server,1:client) */
    char mntpnt[256];          /* mountpoint */
    char user[256];            /* user */
    char passwd[256];          /* password */
    char srctbl[NTRIP_MAXSTR]; /* source table */
    tcpsvr_t* tcp;             /* tcp server */
    ntripc_con_t con[MAXCLI];  /* ntrip client/server connections */
} ntripc_t;
 
typedef struct {             /* udp type */
    int state;               /* state (0:close,1:open) */
    int type;                /* type (0:server,1:client) */
    int port;                /* port */
    char saddr[256];         /* address (server:filter,client:server) */
    struct sockaddr_in addr; /* address resolved */
    socket_t sock;           /* socket descriptor */
} udp_t;
 
typedef struct {         /* ftp download control type */
    int state;           /* state (0:close,1:download,2:complete,3:error) */
    int proto;           /* protocol (0:ftp,1:http) */
    int error;           /* error code (0:no error,1-10:wget error, */
                         /*            11:no temp dir,12:uncompact error) */
    char addr[1024];     /* download address */
    char file[1024];     /* download file path */
    char user[256];      /* user for ftp */
    char passwd[256];    /* password for ftp */
    char local[1024];    /* local file path */
    int topts[4];        /* time options {poff,tint,toff,tretry} (s) */
    gtime_t tnext;       /* next retry time (gpst) */
    rtk_thread_t thread; /* download thread */
} ftp_t;
 
typedef struct {       /* memory buffer type */
    int state, wp, rp; /* state,write/read pointer */
    int bufsize;       /* buffer size (bytes) */
    rtk_lock_t lock;   /* lock flag */
    uint8_t* buf;      /* write buffer */
} membuf_t;
 
/* proto types for static functions ------------------------------------------*/
 
static tcpsvr_t* opentcpsvr(const char* path, char* msg);
static void closetcpsvr(tcpsvr_t* tcpsvr);
static int writetcpsvr(tcpsvr_t* tcpsvr, uint8_t* buff, int n, char* msg);
 
/* global options ------------------------------------------------------------*/
 
static int toinact = 10000;                    /* inactive timeout (ms) */
static int ticonnect = 10000;                  /* interval to re-connect (ms) */
static int tirate = 1000;                      /* averaging time for data rate (ms) */
static int buffsize = 32768;                   /* receive/send buffer size (bytes) */
static char localdir[1024] = "";               /* local directory for ftp/http */
static char proxyaddr[256] = "";               /* http/ntrip/ftp proxy address */
static int ntrip_ver_default = NTRIP_VER_AUTO; /* default ntrip version */
static uint32_t tick_master = 0;               /* time tick master for replay */
static int fswapmargin = 30;                   /* file swap margin (s) */
 
/* open serial ---------------------------------------------------------------*/
static serial_t* openserial(const char* path, int mode, char* msg) {
    const int br[] = {300,
                      600,
                      1200,
                      2400,
                      4800,
                      9600,
                      19200,
                      38400,
                      57600,
                      115200,
                      230400
#ifdef B460800
                      ,
                      460800
#endif
#ifdef B921600
                      ,
                      921600
#endif
    };
    const int nbr = (int)(sizeof(br) / sizeof(br[0]));
    serial_t* serial;
    int i, brate = 9600, bsize = 8, stopb = 1, tcp_port = 0;
    char *p, parity = 'N', dev[128], port[128], fctr[64] = "", path_tcp[32], msg_tcp[128];
    const speed_t bs[] = {B300,
                          B600,
                          B1200,
                          B2400,
                          B4800,
                          B9600,
                          B19200,
                          B38400,
                          B57600,
                          B115200,
                          B230400
#ifdef B460800
                          ,
                          B460800
#endif
#ifdef B921600
                          ,
                          B921600
#endif
    };
    struct termios ios = {0};
    int rw = 0;
    tracet(NULL, 3, "openserial: path=%s mode=%d\n", path, mode);
 
    if (!(serial = (serial_t*)calloc(1, sizeof(serial_t)))) {
        return NULL;
    }
 
    if ((p = strchr(path, ':'))) {
        strncpy(port, path, p - path);
        port[p - path] = '\0';
        sscanf(p, ":%d:%d:%c:%d:%s", &brate, &bsize, &parity, &stopb, fctr);
    } else {
        strcpy(port, path);
    }
 
    if ((p = strchr(path, '#'))) {
        sscanf(p, "#%d", &tcp_port);
    }
    for (i = 0; i < nbr; i++) {
        if (br[i] == brate) {
            break;
        }
    }
    if (i >= nbr) {
        sprintf(msg, "bitrate error (%d)", brate);
        tracet(NULL, 1, "openserial: %s path=%s\n", msg, path);
        free(serial);
        return NULL;
    }
    parity = (char)toupper((int)parity);
 
    sprintf(dev, "/dev/%.*s", (int)sizeof(port) - 6, port);
 
    if ((mode & STR_MODE_R) && (mode & STR_MODE_W)) {
        rw = O_RDWR;
    } else if (mode & STR_MODE_R) {
        rw = O_RDONLY;
    } else if (mode & STR_MODE_W) {
        rw = O_WRONLY;
    }
 
    if ((serial->dev = open(dev, rw | O_NOCTTY | O_NONBLOCK)) < 0) {
        sprintf(msg, "%s open error (%d)", dev, errno);
        tracet(NULL, 1, "openserial: %s dev=%s\n", msg, dev);
        free(serial);
        return NULL;
    }
    tcgetattr(serial->dev, &ios);
    ios.c_iflag = 0;
    ios.c_oflag = 0;
    ios.c_lflag = 0;    /* non-canonical */
    ios.c_cc[VMIN] = 0; /* non-block-mode */
    ios.c_cc[VTIME] = 0;
    cfsetospeed(&ios, bs[i]);
    cfsetispeed(&ios, bs[i]);
    ios.c_cflag |= bsize == 7 ? CS7 : CS8;
    if (parity == 'O') {
        ios.c_cflag |= (PARENB | PARODD);
    } else if (parity == 'E') {
        ios.c_cflag |= PARENB;
    }
    ios.c_cflag |= stopb == 2 ? CSTOPB : 0;
    ios.c_cflag |= !strcmp(fctr, "rts") ? CRTSCTS : 0;
    tcsetattr(serial->dev, TCSANOW, &ios);
    tcflush(serial->dev, TCIOFLUSH);
    sprintf(msg, "%s", dev);
    serial->tcpsvr = NULL;
 
    /* open tcp sever to output received stream */
    if (tcp_port > 0) {
        sprintf(path_tcp, ":%d", tcp_port);
        serial->tcpsvr = opentcpsvr(path_tcp, msg_tcp);
    }
    tracet(NULL, 3, "openserial: dev=%d\n", serial->dev);
    return serial;
}
/* close serial --------------------------------------------------------------*/
static void closeserial(serial_t* serial) {
    tracet(NULL, 3, "closeserial: dev=%d\n", serial->dev);
 
    if (!serial) {
        return;
    }
    close(serial->dev);
    if (serial->tcpsvr) {
        closetcpsvr(serial->tcpsvr);
    }
    free(serial);
}
/* read serial ---------------------------------------------------------------*/
static int readserial(serial_t* serial, uint8_t* buff, int n, char* msg) {
    char msg_tcp[128];
    int nr;
    tracet(NULL, 4, "readserial: dev=%d n=%d\n", serial->dev, n);
    if (!serial) {
        return 0;
    }
    if ((nr = read(serial->dev, buff, n)) < 0) {
        return 0;
    }
    tracet(NULL, 5, "readserial: exit dev=%d nr=%d\n", serial->dev, nr);
 
    /* write received stream to tcp server port */
    if (serial->tcpsvr && nr > 0) {
        writetcpsvr(serial->tcpsvr, buff, (int)nr, msg_tcp);
    }
    return nr;
}
/* write serial --------------------------------------------------------------*/
static int writeserial(serial_t* serial, uint8_t* buff, int n, char* msg) {
    int ns;
 
    if (!serial) {
        return 0;
    }
    tracet(NULL, 3, "writeserial: dev=%d n=%d\n", serial->dev, n);
 
    /* non-blocking port: a full output buffer (EAGAIN) or a short write is not
     * a device error; the unsent part is not resent */
    if ((ns = (int)write(serial->dev, buff, n)) < 0) {
        if (errno != EAGAIN && errno != EWOULDBLOCK) {
            serial->error = 1;
        }
        ns = 0;
    }
    tracet(NULL, 5, "writeserial: exit dev=%d ns=%d\n", serial->dev, ns);
    return ns;
}
/* get state serial ----------------------------------------------------------*/
static int stateserial(serial_t* serial) { return !serial ? 0 : (serial->error ? -1 : 2); }
/* get extended state serial -------------------------------------------------*/
static int statexserial(serial_t* serial, char* msg) {
    char* p = msg;
    int state = !serial ? 0 : (serial->error ? -1 : 2);
 
    p += sprintf(p, "serial:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  dev     = %d\n", serial->dev);
    p += sprintf(p, "  error   = %d\n", serial->error);
    return state;
}
/* open file -----------------------------------------------------------------*/
static int openfile_(file_t* file, gtime_t time, char* msg) {
    FILE* fp;
    double time_sec;
    uint32_t time_time;
    char *rw, tagpath[MAXSTRPATH + 4] = "";
    char tagh[TIMETAGH_LEN + 1] = "";
 
    tracet(NULL, 3, "openfile_: path=%s time=%s\n", file->path, time_str(time, 0));
 
    file->time = utc2gpst(timeget());
    file->tick = file->tick_f = tickget();
    file->fpos_n = 0;
    file->tick_n = 0;
 
    /* use stdin or stdout if file path is null */
    if (!*file->path) {
        file->fp = file->mode & STR_MODE_R ? stdin : stdout;
        return 1;
    }
    /* replace keywords */
    reppath(file->path, file->openpath, time, "", "");
 
    /* create directory */
    if ((file->mode & STR_MODE_W) && !(file->mode & STR_MODE_R)) {
        createdir(file->openpath);
    }
    if (file->mode & STR_MODE_R) {
        rw = "rb";
    } else {
        rw = "wb";
    }
 
    if (!(file->fp = fopen(file->openpath, rw))) {
        sprintf(msg, "file open error: %s", file->openpath);
        tracet(NULL, 1, "openfile: %s\n", msg);
        return 0;
    }
    tracet(NULL, 4, "openfile_: open file %s (%s)\n", file->openpath, rw);
 
    sprintf(tagpath, "%s.tag", file->openpath);
 
    if (file->timetag) { /* output/sync time-tag */
 
        if (!(file->fp_tag = fopen(tagpath, rw))) {
            sprintf(msg, "tag open error: %s", tagpath);
            tracet(NULL, 1, "openfile: %s\n", msg);
            fclose(file->fp);
            return 0;
        }
        tracet(NULL, 4, "openfile_: open tag file %s (%s)\n", tagpath, rw);
 
        if (file->mode & STR_MODE_R) {
            if (fread(&tagh, TIMETAGH_LEN, 1, file->fp_tag) == 1 &&
                fread(&time_time, sizeof(time_time), 1, file->fp_tag) == 1 &&
                fread(&time_sec, sizeof(time_sec), 1, file->fp_tag) == 1) {
                memcpy(&file->tick_f, tagh + TIMETAGH_LEN - 4, sizeof(file->tick_f));
                file->time.time = (time_t)time_time;
                file->time.sec = time_sec;
                file->wtime = file->time;
            } else {
                file->tick_f = 0;
            }
            /* adust time to read playback file */
            timeset(gpst2utc(file->time));
        } else {
            sprintf(tagh, "TIMETAG %s %s", MRTKLIB_SOFTNAME, MRTKLIB_VERSION_STRING);
            memcpy(tagh + TIMETAGH_LEN - 4, &file->tick_f, sizeof(file->tick_f));
            time_time = (uint32_t)file->time.time;
            time_sec = file->time.sec;
            fwrite(&tagh, 1, TIMETAGH_LEN, file->fp_tag);
            fwrite(&time_time, 1, sizeof(time_time), file->fp_tag);
            fwrite(&time_sec, 1, sizeof(time_sec), file->fp_tag);
            /* time tag file structure   */
            /*   HEADER(60)+TICK(4)+TIME(4+8)+ */
            /*   TICK0(4)+FPOS0(4/8)+    */
            /*   TICK1(4)+FPOS1(4/8)+... */
        }
    } else if (file->mode & STR_MODE_W) { /* remove time-tag */
        if ((fp = fopen(tagpath, "rb"))) {
            fclose(fp);
            remove(tagpath);
        }
    }
    return 1;
}
/* close file ----------------------------------------------------------------*/
static void closefile_(file_t* file) {
    tracet(NULL, 3, "closefile_: path=%s\n", file->path);
 
    if (file->fp) {
        fclose(file->fp);
    }
    if (file->fp_tag) {
        fclose(file->fp_tag);
    }
    if (file->fp_tmp) {
        fclose(file->fp_tmp);
    }
    if (file->fp_tag_tmp) {
        fclose(file->fp_tag_tmp);
    }
    file->fp = file->fp_tag = file->fp_tmp = file->fp_tag_tmp = NULL;
 
    /* reset time offset */
    timereset();
}
/* open file (path=filepath[::T[::+<off>][::x<speed>]][::S=swapintv][::P={4|8}] */
static file_t* openfile(const char* path, int mode, char* msg) {
    file_t* file;
    gtime_t time, time0 = {0};
    double speed = 1.0, start = 0.0, swapintv = 0.0;
    char* p;
    int timetag = 0, size_fpos = 4; /* default 4B */
 
    tracet(NULL, 3, "openfile: path=%s mode=%d\n", path, mode);
 
    if (!(mode & (STR_MODE_R | STR_MODE_W))) {
        return NULL;
    }
 
    /* file options */
    for (p = (char*)path; (p = strstr(p, "::")); p += 2) { /* file options */
        if (*(p + 2) == 'T') {
            timetag = 1;
        } else if (*(p + 2) == '+') {
            sscanf(p + 2, "+%lf", &start);
        } else if (*(p + 2) == 'x') {
            sscanf(p + 2, "x%lf", &speed);
        } else if (*(p + 2) == 'S') {
            sscanf(p + 2, "S=%lf", &swapintv);
        } else if (*(p + 2) == 'P') {
            sscanf(p + 2, "P=%d", &size_fpos);
        }
    }
    if (start <= 0.0) {
        start = 0.0;
    }
    if (swapintv <= 0.0) {
        swapintv = 0.0;
    }
 
    if (!(file = (file_t*)malloc(sizeof(file_t)))) {
        return NULL;
    }
 
    file->fp = file->fp_tag = file->fp_tmp = file->fp_tag_tmp = NULL;
    strcpy(file->path, path);
    if ((p = strstr(file->path, "::"))) {
        *p = '\0';
    }
    file->openpath[0] = '\0';
    file->mode = mode;
    file->timetag = timetag;
    file->repmode = 0;
    file->offset = 0;
    file->size_fpos = size_fpos;
    file->time = file->wtime = time0;
    file->tick = file->tick_f = file->tick_n = file->fpos_n = 0;
    file->start = start;
    file->speed = speed;
    file->swapintv = swapintv;
    rtk_initlock(&file->lock);
 
    time = utc2gpst(timeget());
 
    /* open new file */
    if (!openfile_(file, time, msg)) {
        free(file);
        return NULL;
    }
    return file;
}
/* close file ----------------------------------------------------------------*/
static void closefile(file_t* file) {
    tracet(NULL, 3, "closefile: fp=%d\n", file->fp);
 
    if (!file) {
        return;
    }
    closefile_(file);
    free(file);
}
/* open new swap file --------------------------------------------------------*/
static void swapfile(file_t* file, gtime_t time, char* msg) {
    char openpath[MAXSTRPATH];
 
    tracet(NULL, 3, "swapfile: fp=%d time=%s\n", file->fp, time_str(time, 0));
 
    /* return if old swap file open */
    if (file->fp_tmp || file->fp_tag_tmp) {
        return;
    }
 
    /* check path of new swap file */
    reppath(file->path, openpath, time, "", "");
 
    if (!strcmp(openpath, file->openpath)) {
        tracet(NULL, 2, "swapfile: no need to swap %s\n", openpath);
        return;
    }
    /* save file pointer to temporary pointer */
    file->fp_tmp = file->fp;
    file->fp_tag_tmp = file->fp_tag;
 
    /* open new swap file */
    openfile_(file, time, msg);
}
/* close old swap file -------------------------------------------------------*/
static void swapclose(file_t* file) {
    tracet(NULL, 3, "swapclose: fp_tmp=%d\n", file->fp_tmp);
 
    if (file->fp_tmp) {
        fclose(file->fp_tmp);
    }
    if (file->fp_tag_tmp) {
        fclose(file->fp_tag_tmp);
    }
    file->fp_tmp = file->fp_tag_tmp = NULL;
}
/* get state file ------------------------------------------------------------*/
static int statefile(file_t* file) { return file ? 2 : 0; }
/* get extended state file ---------------------------------------------------*/
static int statexfile(file_t* file, char* msg) {
    char *p = msg, tstr1[32], tstr2[32];
    int state = file ? 2 : 0;
 
    p += sprintf(p, "file:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    time2str(file->time, tstr1, 3);
    time2str(file->wtime, tstr2, 3);
    p += sprintf(p, "  path    = %s\n", file->path);
    p += sprintf(p, "  openpath= %s\n", file->openpath);
    p += sprintf(p, "  mode    = %d\n", file->mode);
    p += sprintf(p, "  timetag = %d\n", file->timetag);
    p += sprintf(p, "  repmode = %d\n", file->repmode);
    p += sprintf(p, "  offsete = %d\n", file->offset);
    p += sprintf(p, "  time    = %s\n", tstr1);
    p += sprintf(p, "  wtime   = %s\n", tstr2);
    p += sprintf(p, "  tick    = %u\n", file->tick);
    p += sprintf(p, "  tick_f  = %u\n", file->tick_f);
    p += sprintf(p, "  start   = %.3f\n", file->start);
    p += sprintf(p, "  speed   = %.3f\n", file->speed);
    p += sprintf(p, "  swapintv= %.3f\n", file->swapintv);
    return state;
}
/* read file -----------------------------------------------------------------*/
static int readfile(file_t* file, uint8_t* buff, int nmax, char* msg) {
    struct timeval tv = {0};
    fd_set rs;
    uint64_t fpos_8B;
    uint32_t t, tick, fpos_4B;
    long pos, n;
    int nr = 0;
 
    tracet(NULL, 4, "readfile: fp=%d nmax=%d\n", file->fp, nmax);
 
    if (!file) {
        return 0;
    }
 
    if (file->fp == stdin) {
        /* input from stdin */
        FD_ZERO(&rs);
        FD_SET(0, &rs);
        if (!select(1, &rs, NULL, NULL, &tv)) {
            return 0;
        }
        if ((nr = (int)read(0, buff, nmax)) < 0) {
            return 0;
        }
        return nr;
    }
    if (file->fp_tag) {
        /* target tick */
        if (file->repmode) { /* slave */
            t = (uint32_t)(tick_master + file->offset);
        } else { /* master */
            t = (uint32_t)((tickget() - file->tick) * file->speed + file->start * 1000.0);
            tick_master = t;
        }
        /* seek time-tag file to get next tick and file position */
        while ((int)(file->tick_n - t) <= 0) {
            if (fread(&file->tick_n, sizeof(tick), 1, file->fp_tag) < 1 ||
                fread((file->size_fpos == 4) ? (void*)&fpos_4B : (void*)&fpos_8B, file->size_fpos, 1, file->fp_tag) <
                    1) {
                file->tick_n = (uint32_t)(-1);
                pos = ftell(file->fp);
                fseek(file->fp, 0L, SEEK_END);
                file->fpos_n = ftell(file->fp);
                fseek(file->fp, pos, SEEK_SET);
                break;
            }
            file->fpos_n = (long)((file->size_fpos == 4) ? fpos_4B : fpos_8B);
        }
        if (file->tick_n == (uint32_t)(-1)) {
            sprintf(msg, "end");
        } else {
            sprintf(msg, "T%+.1fs", (int)t * 0.001);
            file->wtime = timeadd(file->time, (int)t * 0.001);
            timeset(timeadd(gpst2utc(file->time), (int)file->tick_n * 0.001));
        }
        if ((n = file->fpos_n - ftell(file->fp)) < nmax) {
            nmax = n;
        }
    }
    if (nmax > 0) {
        nr = (int)fread(buff, 1, nmax, file->fp);
    }
    if (feof(file->fp)) {
        sprintf(msg, "end");
    }
    tracet(NULL, 5, "readfile: fp=%d nr=%d\n", file->fp, nr);
    return nr;
}
/* write file ----------------------------------------------------------------*/
static int writefile(file_t* file, uint8_t* buff, int n, char* msg) {
    gtime_t wtime;
    uint64_t fpos_8B;
    uint32_t tick = tickget(), fpos_4B;
    int week1, week2, ns;
    double tow1, tow2, intv;
    long fpos, fpos_tmp = 0;
 
    tracet(NULL, 4, "writefile: fp=%d n=%d\n", file->fp, n);
 
    if (!file) {
        return 0;
    }
 
    wtime = utc2gpst(timeget()); /* write time in gpst */
 
    /* swap writing file */
    if (file->swapintv > 0.0 && file->wtime.time != 0) {
        intv = file->swapintv * 3600.0;
        tow1 = time2gpst(file->wtime, &week1);
        tow2 = time2gpst(wtime, &week2);
        tow2 += 604800.0 * (week2 - week1);
 
        /* open new swap file */
        if (floor((tow1 + fswapmargin) / intv) < floor((tow2 + fswapmargin) / intv)) {
            swapfile(file, timeadd(wtime, fswapmargin), msg);
        }
        /* close old swap file */
        if (floor((tow1 - fswapmargin) / intv) < floor((tow2 - fswapmargin) / intv)) {
            swapclose(file);
        }
    }
    if (!file->fp) {
        return 0;
    }
 
    ns = (int)fwrite(buff, 1, n, file->fp);
    fpos = ftell(file->fp);
    fflush(file->fp);
    file->wtime = wtime;
 
    if (file->fp_tmp) {
        fwrite(buff, 1, n, file->fp_tmp);
        fpos_tmp = ftell(file->fp_tmp);
        fflush(file->fp_tmp);
    }
    if (file->fp_tag) {
        tick -= file->tick;
        fwrite(&tick, 1, sizeof(tick), file->fp_tag);
        if (file->size_fpos == 4) {
            fpos_4B = (uint32_t)fpos;
            fwrite(&fpos_4B, 1, sizeof(fpos_4B), file->fp_tag);
        } else {
            fpos_8B = (uint64_t)fpos;
            fwrite(&fpos_8B, 1, sizeof(fpos_8B), file->fp_tag);
        }
        fflush(file->fp_tag);
 
        if (file->fp_tag_tmp) {
            fwrite(&tick, 1, sizeof(tick), file->fp_tag_tmp);
            if (file->size_fpos == 4) {
                fpos_4B = (uint32_t)fpos_tmp;
                fwrite(&fpos_4B, 1, sizeof(fpos_4B), file->fp_tag_tmp);
            } else {
                fpos_8B = (uint64_t)fpos_tmp;
                fwrite(&fpos_8B, 1, sizeof(fpos_8B), file->fp_tag_tmp);
            }
            fflush(file->fp_tag_tmp);
        }
    }
    tracet(NULL, 5, "writefile: fp=%d ns=%d tick=%5d fpos=%d\n", file->fp, ns, tick, fpos);
 
    return ns;
}
/* sync files by time-tag ----------------------------------------------------*/
static void syncfile(file_t* file1, file_t* file2) {
    if (!file1->fp_tag || !file2->fp_tag) {
        return;
    }
    file1->repmode = 0;
    file2->repmode = 1;
    file2->offset = (int)(file1->tick_f - file2->tick_f);
}
/* decode percent-encoded characters in-place (%XX -> char) -----------------*/
static void urldecode(char* str) {
    char *src = str, *dst = str;
    while (*src) {
        if (*src == '%' && src[1] && src[2]) {
            int hi = src[1], lo = src[2];
            hi = (hi >= '0' && hi <= '9')   ? hi - '0'
                 : (hi >= 'A' && hi <= 'F') ? hi - 'A' + 10
                 : (hi >= 'a' && hi <= 'f') ? hi - 'a' + 10
                                            : -1;
            lo = (lo >= '0' && lo <= '9')   ? lo - '0'
                 : (lo >= 'A' && lo <= 'F') ? lo - 'A' + 10
                 : (lo >= 'a' && lo <= 'f') ? lo - 'a' + 10
                                            : -1;
            if (hi >= 0 && lo >= 0) {
                char ch = (char)((hi << 4) | lo);
                if (ch == '\0' || ch == '\r' || ch == '\n' || ch == ' ') {
                    /* reject NUL, CR, LF, space to prevent string truncation,
                     * request injection, and v1 SOURCE command delimiter issues */
                    src += 3;
                    continue;
                }
                *dst++ = ch;
                src += 3;
                continue;
            }
        }
        *dst++ = *src++;
    }
    *dst = '\0';
}
/* decode tcp/ntrip path (path=[user[:passwd]@]addr[:port][/mntpnt[:str]]) ---*/
static void decodetcppath(const char* path, char* addr, char* port, char* user, char* passwd, char* mntpnt, char* str) {
    char buff[MAXSTRPATH], *p, *q;
 
    tracet(NULL, 4, "decodetcpepath: path=%s\n", path);
 
    if (port) {
        *port = '\0';
    }
    if (user) {
        *user = '\0';
    }
    if (passwd) {
        *passwd = '\0';
    }
    if (mntpnt) {
        *mntpnt = '\0';
    }
    if (str) {
        *str = '\0';
    }
 
    strcpy(buff, path);
 
    if (!(p = strrchr(buff, '@'))) {
        p = buff;
    }
 
    if ((p = strchr(p, '/'))) {
        if ((q = strchr(p + 1, ':'))) {
            *q = '\0';
            if (str) {
                sprintf(str, "%.*s", NTRIP_MAXSTR - 1, q + 1);
            }
        }
        *p = '\0';
        if (mntpnt) {
            sprintf(mntpnt, "%.255s", p + 1);
        }
    }
    if ((p = strrchr(buff, '@'))) {
        *p++ = '\0';
        if ((q = strchr(buff, ':'))) {
            *q = '\0';
            if (passwd) {
                sprintf(passwd, "%.255s", q + 1);
            }
        }
        if (user) {
            sprintf(user, "%.255s", buff);
            urldecode(user);
        }
        if (passwd) {
            urldecode(passwd);
        }
    } else {
        p = buff;
    }
 
    if ((q = strchr(p, ':'))) {
        *q = '\0';
        if (port) {
            sprintf(port, "%.255s", q + 1);
        }
    }
    if (addr) {
        sprintf(addr, "%.255s", p);
    }
}
/* get socket error ----------------------------------------------------------*/
static int errsock(void) { return errno; }
 
/* set socket option ---------------------------------------------------------*/
static int setsock(socket_t sock, char* msg) {
    int bs = buffsize, mode = 1;
    struct timeval tv = {0};
    tracet(NULL, 3, "setsock: sock=%d\n", sock);
 
    if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv)) == -1 ||
        setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (const char*)&tv, sizeof(tv)) == -1) {
        sprintf(msg, "sockopt error: notimeo");
        tracet(NULL, 1, "setsock: setsockopt error 1 sock=%d err=%d\n", sock, errsock());
        closesocket(sock);
        return 0;
    }
    if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (const char*)&bs, sizeof(bs)) == -1 ||
        setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (const char*)&bs, sizeof(bs)) == -1) {
        tracet(NULL, 1, "setsock: setsockopt error 2 sock=%d err=%d bs=%d\n", sock, errsock(), bs);
        sprintf(msg, "sockopt error: bufsiz");
    }
    if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (const char*)&mode, sizeof(mode)) == -1) {
        tracet(NULL, 1, "setsock: setsockopt error 3 sock=%d err=%d\n", sock, errsock());
        sprintf(msg, "sockopt error: nodelay");
    }
    return 1;
}
/* non-block accept ----------------------------------------------------------*/
static socket_t accept_nb(socket_t sock, struct sockaddr* addr, socklen_t* len) {
    struct timeval tv = {0};
    fd_set rs;
    int ret;
 
    FD_ZERO(&rs);
    FD_SET(sock, &rs);
    ret = select(sock + 1, &rs, NULL, NULL, &tv);
    if (ret <= 0) {
        return (socket_t)ret;
    }
    return accept(sock, addr, len);
}
/* non-block connect ---------------------------------------------------------*/
static int connect_nb(socket_t sock, struct sockaddr* addr, socklen_t len) {
    struct timeval tv = {0};
    fd_set rs, ws;
    int err, flag;
 
    flag = fcntl(sock, F_GETFL, 0);
    fcntl(sock, F_SETFL, flag | O_NONBLOCK);
    if (connect(sock, addr, len) == -1) {
        err = errsock();
        if (err != EISCONN && err != EINPROGRESS && err != EALREADY) {
            return -1;
        }
        FD_ZERO(&rs);
        FD_SET(sock, &rs);
        ws = rs;
        if (select(sock + 1, &rs, &ws, NULL, &tv) == 0) {
            return 0;
        }
    }
    return 1;
}
/* non-block receive ---------------------------------------------------------*/
static int recv_nb(socket_t sock, uint8_t* buff, int n) {
    struct timeval tv = {0};
    fd_set rs;
    int ret, nr;
 
    FD_ZERO(&rs);
    FD_SET(sock, &rs);
    ret = select(sock + 1, &rs, NULL, NULL, &tv);
    if (ret <= 0) {
        return ret;
    }
    nr = recv(sock, (char*)buff, n, 0);
    return nr <= 0 ? -1 : nr;
}
/* non-block send ------------------------------------------------------------*/
static int send_nb(socket_t sock, uint8_t* buff, int n) {
    struct timeval tv = {0};
    fd_set ws;
    int ret, ns;
 
    FD_ZERO(&ws);
    FD_SET(sock, &ws);
    ret = select(sock + 1, NULL, &ws, NULL, &tv);
    if (ret <= 0) {
        return ret;
    }
    ns = send(sock, (char*)buff, n, 0);
    return ns < n ? -1 : ns;
}
/* generate tcp socket -------------------------------------------------------*/
static int gentcp(tcp_t* tcp, int type, char* msg) {
    struct hostent* hp;
#ifdef SVR_REUSEADDR
    int opt = 1;
#endif
 
    tracet(NULL, 3, "gentcp: type=%d\n", type);
 
    /* generate socket */
    if ((tcp->sock = socket(AF_INET, SOCK_STREAM, 0)) == (socket_t)-1) {
        sprintf(msg, "socket error (%d)", errsock());
        tracet(NULL, 1, "gentcp: socket error err=%d\n", errsock());
        tcp->state = -1;
        return 0;
    }
    if (!setsock(tcp->sock, msg)) {
        tcp->state = -1;
        return 0;
    }
    memset(&tcp->addr, 0, sizeof(tcp->addr));
    tcp->addr.sin_family = AF_INET;
    tcp->addr.sin_port = htons(tcp->port);
 
    if (type == 0) { /* server socket */
 
#ifdef SVR_REUSEADDR
        /* multiple-use of server socket */
        setsockopt(tcp->sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
#endif
        if (bind(tcp->sock, (struct sockaddr*)&tcp->addr, sizeof(tcp->addr)) == -1) {
            sprintf(msg, "bind error (%d) : %d", errsock(), tcp->port);
            tracet(NULL, 1, "gentcp: bind error port=%d err=%d\n", tcp->port, errsock());
            closesocket(tcp->sock);
            tcp->state = -1;
            return 0;
        }
        listen(tcp->sock, 5);
    } else { /* client socket */
        if (!(hp = gethostbyname(tcp->saddr))) {
            sprintf(msg, "address error (%s)", tcp->saddr);
            tracet(NULL, 1, "gentcp: gethostbyname error addr=%s err=%d\n", tcp->saddr, errsock());
            closesocket(tcp->sock);
            tcp->state = 0;
            tcp->tcon = ticonnect;
            tcp->tdis = tickget();
            return 0;
        }
        memcpy(&tcp->addr.sin_addr, hp->h_addr_list[0], hp->h_length);
    }
    tcp->state = 1;
    tcp->tact = tickget();
    tracet(NULL, 5, "gentcp: exit sock=%d\n", tcp->sock);
    return 1;
}
/* disconnect tcp ------------------------------------------------------------*/
static void discontcp(tcp_t* tcp, int tcon) {
    tracet(NULL, 3, "discontcp: sock=%d tcon=%d\n", tcp->sock, tcon);
 
    closesocket(tcp->sock);
    tcp->state = 0;
    tcp->tcon = tcon;
    tcp->tdis = tickget();
}
/* open tcp server -----------------------------------------------------------*/
static tcpsvr_t* opentcpsvr(const char* path, char* msg) {
    tcpsvr_t *tcpsvr, tcpsvr0 = {{0}};
    char port[256] = "";
 
    tracet(NULL, 3, "opentcpsvr: path=%s\n", path);
 
    if (!(tcpsvr = (tcpsvr_t*)malloc(sizeof(tcpsvr_t)))) {
        return NULL;
    }
    *tcpsvr = tcpsvr0;
    decodetcppath(path, tcpsvr->svr.saddr, port, NULL, NULL, NULL, NULL);
    if (sscanf(port, "%d", &tcpsvr->svr.port) < 1) {
        sprintf(msg, "port error: %s", port);
        tracet(NULL, 1, "opentcpsvr: port error port=%s\n", port);
        free(tcpsvr);
        return NULL;
    }
    if (!gentcp(&tcpsvr->svr, 0, msg)) {
        free(tcpsvr);
        return NULL;
    }
    tcpsvr->svr.tcon = 0;
    return tcpsvr;
}
/* close tcp server ----------------------------------------------------------*/
static void closetcpsvr(tcpsvr_t* tcpsvr) {
    int i;
 
    tracet(NULL, 3, "closetcpsvr:\n");
 
    for (i = 0; i < MAXCLI; i++) {
        if (tcpsvr->cli[i].state) {
            closesocket(tcpsvr->cli[i].sock);
        }
    }
    closesocket(tcpsvr->svr.sock);
    free(tcpsvr);
}
/* update tcp server ---------------------------------------------------------*/
static void updatetcpsvr(tcpsvr_t* tcpsvr, char* msg) {
    char saddr[256] = "";
    int i, n = 0;
 
    tracet(NULL, 4, "updatetcpsvr: state=%d\n", tcpsvr->svr.state);
 
    if (tcpsvr->svr.state == 0) {
        return;
    }
 
    for (i = 0; i < MAXCLI; i++) {
        if (!tcpsvr->cli[i].state) {
            continue;
        }
        strcpy(saddr, tcpsvr->cli[i].saddr);
        n++;
    }
    if (n == 0) {
        tcpsvr->svr.state = 1;
        sprintf(msg, "waiting...");
        return;
    }
    tcpsvr->svr.state = 2;
    if (n == 1) {
        sprintf(msg, "%s", saddr);
    } else {
        sprintf(msg, "%d clients", n);
    }
}
/* accept client connection --------------------------------------------------*/
static int accsock(tcpsvr_t* tcpsvr, char* msg) {
    struct sockaddr_in addr;
    socket_t sock;
    socklen_t len = sizeof(addr);
    int i, err;
 
    tracet(NULL, 4, "accsock: sock=%d\n", tcpsvr->svr.sock);
 
    for (i = 0; i < MAXCLI; i++) {
        if (tcpsvr->cli[i].state == 0) {
            break;
        }
    }
    if (i >= MAXCLI) {
        tracet(NULL, 2, "accsock: too many clients sock=%d\n", tcpsvr->svr.sock);
        return 0;
    }
    if ((sock = accept_nb(tcpsvr->svr.sock, (struct sockaddr*)&addr, &len)) == (socket_t)-1) {
        err = errsock();
        sprintf(msg, "accept error (%d)", err);
        tracet(NULL, 1, "accsock: accept error sock=%d err=%d\n", tcpsvr->svr.sock, err);
        closesocket(tcpsvr->svr.sock);
        tcpsvr->svr.state = 0;
        return 0;
    }
    if (sock == 0) {
        return 0;
    }
    if (!setsock(sock, msg)) {
        return 0;
    }
 
    tcpsvr->cli[i].sock = sock;
    memcpy(&tcpsvr->cli[i].addr, &addr, sizeof(addr));
    strcpy(tcpsvr->cli[i].saddr, inet_ntoa(addr.sin_addr));
    sprintf(msg, "%s", tcpsvr->cli[i].saddr);
    tracet(NULL, 3, "accsock: connected sock=%d addr=%s i=%d\n", tcpsvr->cli[i].sock, tcpsvr->cli[i].saddr, i);
    tcpsvr->cli[i].state = 2;
    tcpsvr->cli[i].tact = tickget();
    return 1;
}
/* wait socket accept --------------------------------------------------------*/
static int waittcpsvr(tcpsvr_t* tcpsvr, char* msg) {
    tracet(NULL, 4, "waittcpsvr: sock=%d state=%d\n", tcpsvr->svr.sock, tcpsvr->svr.state);
 
    if (tcpsvr->svr.state <= 0) {
        return 0;
    }
 
    while (accsock(tcpsvr, msg)) {
        /* accept all pending connections */
    }
 
    updatetcpsvr(tcpsvr, msg);
    return tcpsvr->svr.state == 2;
}
/* read tcp server -----------------------------------------------------------*/
static int readtcpsvr(tcpsvr_t* tcpsvr, uint8_t* buff, int n, char* msg) {
    int i, nr, err;
 
    tracet(NULL, 4, "readtcpsvr: state=%d\n", tcpsvr->svr.state);
 
    if (!waittcpsvr(tcpsvr, msg)) {
        return 0;
    }
 
    for (i = 0; i < MAXCLI; i++) {
        if (tcpsvr->cli[i].state != 2) {
            continue;
        }
 
        if ((nr = recv_nb(tcpsvr->cli[i].sock, buff, n)) == -1) {
            if ((err = errsock())) {
                tracet(NULL, 2, "readtcpsvr: recv error sock=%d err=%d\n", tcpsvr->cli[i].sock, err);
            }
            discontcp(&tcpsvr->cli[i], ticonnect);
            updatetcpsvr(tcpsvr, msg);
        }
        if (nr > 0) {
            tcpsvr->cli[i].tact = tickget();
            return nr;
        }
    }
    return 0;
}
/* write tcp server ----------------------------------------------------------*/
static int writetcpsvr(tcpsvr_t* tcpsvr, uint8_t* buff, int n, char* msg) {
    int i, ns = 0, nmax = 0, err;
 
    tracet(NULL, 4, "writetcpsvr: state=%d n=%d\n", tcpsvr->svr.state, n);
 
    if (!waittcpsvr(tcpsvr, msg)) {
        return 0;
    }
 
    for (i = 0; i < MAXCLI; i++) {
        if (tcpsvr->cli[i].state != 2) {
            continue;
        }
 
        if ((ns = send_nb(tcpsvr->cli[i].sock, buff, n)) == -1) {
            if ((err = errsock())) {
                tracet(NULL, 2, "writetcpsvr: send error i=%d sock=%d err=%d\n", i, tcpsvr->cli[i].sock, err);
            }
            discontcp(&tcpsvr->cli[i], ticonnect);
            updatetcpsvr(tcpsvr, msg);
        } else {
            if (ns > nmax) {
                nmax = ns;
            }
            if (ns > 0) {
                tcpsvr->cli[i].tact = tickget();
            }
        }
    }
    return nmax;
}
/* get state tcp server ------------------------------------------------------*/
static int statetcpsvr(tcpsvr_t* tcpsvr) { return tcpsvr ? tcpsvr->svr.state : 0; }
/* print extended state tcp --------------------------------------------------*/
static int statextcp(tcp_t* tcp, char* msg) {
    char* p = msg;
 
    p += sprintf(p, "    state = %d\n", tcp->state);
    p += sprintf(p, "    saddr = %s\n", tcp->saddr);
    p += sprintf(p, "    port  = %d\n", tcp->port);
    p += sprintf(p, "    sock  = %d\n", (int)tcp->sock);
#if 0 /* for debug */
    p+=sprintf(p,"    tcon  = %d\n",tcp->tcon);
    p+=sprintf(p,"    tact  = %u\n",tcp->tact);
    p+=sprintf(p,"    tdis  = %u\n",tcp->tdis);
#endif
    return (int)(p - msg);
}
/* get extended state tcp server ---------------------------------------------*/
static int statextcpsvr(tcpsvr_t* tcpsvr, char* msg) {
    char* p = msg;
    int i, state = tcpsvr ? tcpsvr->svr.state : 0;
 
    p += sprintf(p, "tcpsvr:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  svr:\n");
    p += statextcp(&tcpsvr->svr, p);
    for (i = 0; i < MAXCLI; i++) {
        if (!tcpsvr->cli[i].state) {
            continue;
        }
        p += sprintf(p, "  cli#%d:\n", i);
        p += statextcp(tcpsvr->cli + i, p);
    }
    return state;
}
/* connect server ------------------------------------------------------------*/
static int consock(tcpcli_t* tcpcli, char* msg) {
    int stat, err;
 
    tracet(NULL, 4, "consock: sock=%d\n", tcpcli->svr.sock);
 
    /* wait re-connect */
    if (tcpcli->svr.tcon < 0 || (tcpcli->svr.tcon > 0 && (int)(tickget() - tcpcli->svr.tdis) < tcpcli->svr.tcon)) {
        return 0;
    }
    /* non-block connect */
    if ((stat = connect_nb(tcpcli->svr.sock, (struct sockaddr*)&tcpcli->svr.addr, sizeof(tcpcli->svr.addr))) == -1) {
        err = errsock();
        sprintf(msg, "connect error (%d)", err);
        tracet(NULL, 2, "consock: connect error sock=%d err=%d\n", tcpcli->svr.sock, err);
        closesocket(tcpcli->svr.sock);
        tcpcli->svr.state = 0;
        return 0;
    }
    if (!stat) { /* not connect */
        sprintf(msg, "connecting...");
        return 0;
    }
    sprintf(msg, "%s", tcpcli->svr.saddr);
    tracet(NULL, 3, "consock: connected sock=%d addr=%s\n", tcpcli->svr.sock, tcpcli->svr.saddr);
    tcpcli->svr.state = 2;
    tcpcli->svr.tact = tickget();
    return 1;
}
/* open tcp client -----------------------------------------------------------*/
static tcpcli_t* opentcpcli(const char* path, char* msg) {
    tcpcli_t *tcpcli, tcpcli0 = {{0}};
    char port[256] = "";
 
    tracet(NULL, 3, "opentcpcli: path=%s\n", path);
 
    if (!(tcpcli = (tcpcli_t*)malloc(sizeof(tcpcli_t)))) {
        return NULL;
    }
    *tcpcli = tcpcli0;
    decodetcppath(path, tcpcli->svr.saddr, port, NULL, NULL, NULL, NULL);
    if (sscanf(port, "%d", &tcpcli->svr.port) < 1) {
        sprintf(msg, "port error: %s", port);
        tracet(NULL, 2, "opentcp: port error port=%s\n", port);
        free(tcpcli);
        return NULL;
    }
    tcpcli->svr.tcon = 0;
    tcpcli->toinact = toinact;
    tcpcli->tirecon = ticonnect;
    return tcpcli;
}
/* close tcp client ----------------------------------------------------------*/
static void closetcpcli(tcpcli_t* tcpcli) {
    tracet(NULL, 3, "closetcpcli: sock=%d\n", tcpcli->svr.sock);
 
    closesocket(tcpcli->svr.sock);
    free(tcpcli);
}
/* wait socket connect -------------------------------------------------------*/
static int waittcpcli(tcpcli_t* tcpcli, char* msg) {
    tracet(NULL, 4, "waittcpcli: sock=%d state=%d\n", tcpcli->svr.sock, tcpcli->svr.state);
 
    if (tcpcli->svr.state < 0) {
        return 0;
    }
 
    if (tcpcli->svr.state == 0) { /* close */
        if (!gentcp(&tcpcli->svr, 1, msg)) {
            return 0;
        }
    }
    if (tcpcli->svr.state == 1) { /* wait */
        if (!consock(tcpcli, msg)) {
            return 0;
        }
    }
    if (tcpcli->svr.state == 2) { /* connect */
        if (tcpcli->toinact > 0 && (int)(tickget() - tcpcli->svr.tact) > tcpcli->toinact) {
            sprintf(msg, "timeout");
            tracet(NULL, 2, "waittcpcli: inactive timeout sock=%d\n", tcpcli->svr.sock);
            discontcp(&tcpcli->svr, tcpcli->tirecon);
            return 0;
        }
    }
    return 1;
}
/* read tcp client -----------------------------------------------------------*/
static int readtcpcli(tcpcli_t* tcpcli, uint8_t* buff, int n, char* msg) {
    int nr, err;
 
    tracet(NULL, 4, "readtcpcli: sock=%d\n", tcpcli->svr.sock);
 
    if (!waittcpcli(tcpcli, msg)) {
        return 0;
    }
 
    if ((nr = recv_nb(tcpcli->svr.sock, buff, n)) == -1) {
        if ((err = errsock())) {
            tracet(NULL, 2, "readtcpcli: recv error sock=%d err=%d\n", tcpcli->svr.sock, err);
            sprintf(msg, "recv error (%d)", err);
        } else {
            sprintf(msg, "disconnected");
        }
        discontcp(&tcpcli->svr, tcpcli->tirecon);
        return 0;
    }
    if (nr > 0) {
        tcpcli->svr.tact = tickget();
    }
    tracet(NULL, 5, "readtcpcli: exit sock=%d nr=%d\n", tcpcli->svr.sock, nr);
    return nr;
}
/* write tcp client ----------------------------------------------------------*/
static int writetcpcli(tcpcli_t* tcpcli, uint8_t* buff, int n, char* msg) {
    int ns, err;
 
    tracet(NULL, 3, "writetcpcli: sock=%d state=%d n=%d\n", tcpcli->svr.sock, tcpcli->svr.state, n);
 
    if (!waittcpcli(tcpcli, msg)) {
        return 0;
    }
 
    if ((ns = send_nb(tcpcli->svr.sock, buff, n)) == -1) {
        if ((err = errsock())) {
            tracet(NULL, 2, "writetcp: send error sock=%d err=%d\n", tcpcli->svr.sock, err);
            sprintf(msg, "send error (%d)", err);
        }
        discontcp(&tcpcli->svr, tcpcli->tirecon);
        return 0;
    }
    if (ns > 0) {
        tcpcli->svr.tact = tickget();
    }
    tracet(NULL, 5, "writetcpcli: exit sock=%d ns=%d\n", tcpcli->svr.sock, ns);
    return ns;
}
/* get state tcp client ------------------------------------------------------*/
static int statetcpcli(tcpcli_t* tcpcli) { return tcpcli ? tcpcli->svr.state : 0; }
/* get extended state tcp client ---------------------------------------------*/
static int statextcpcli(tcpcli_t* tcpcli, char* msg) { return tcpcli ? tcpcli->svr.state : 0; }
/* base64 encoder ------------------------------------------------------------*/
static int encbase64(char* str, const uint8_t* byte, int n) {
    const char table[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    int i, j, k, b;
 
    tracet(NULL, 4, "encbase64: n=%d\n", n);
 
    for (i = j = 0; i / 8 < n;) {
        for (k = b = 0; k < 6; k++, i++) {
            b <<= 1;
            if (i / 8 < n) {
                b |= (byte[i / 8] >> (7 - i % 8)) & 0x1;
            }
        }
        str[j++] = table[b];
    }
    while (j & 0x3) {
        str[j++] = '=';
    }
    str[j] = '\0';
    tracet(NULL, 5, "encbase64: str=%s\n", str);
    return j;
}
/* safe append to buffer with overflow tracking */
#define BUFADD(p, rem, ...)                     \
    do {                                        \
        int _n = snprintf(p, rem, __VA_ARGS__); \
        if (_n > 0 && _n < (rem)) {             \
            (p) += _n;                          \
            (rem) -= _n;                        \
        } else {                                \
            (rem) = 0;                          \
        }                                       \
    } while (0)
/* send ntrip server request -------------------------------------------------*/
static int reqntrip_s(ntrip_t* ntrip, char* msg) {
    char buff[2048], user[514], *p = buff;
    int rem = (int)sizeof(buff);
    int use_v2 = (ntrip->ver == NTRIP_VER_2 ||
                  (ntrip->ver == NTRIP_VER_AUTO && (!ntrip->v2_tried || ntrip->ver_neg == NTRIP_VER_2)));
 
    tracet(NULL, 3, "reqntrip_s: state=%d ver=%d\n", ntrip->state, ntrip->ver);
 
    if (use_v2) {
        /* NTRIP v2: POST with HTTP/1.1 (use absolute URL for proxy) */
        if (*ntrip->url) {
            BUFADD(p, rem, "POST %s/%s HTTP/1.1\r\n", ntrip->url, ntrip->mntpnt);
        } else {
            BUFADD(p, rem, "POST /%s HTTP/1.1\r\n", ntrip->mntpnt);
        }
        BUFADD(p, rem, "Host: %s\r\n", ntrip->host);
        BUFADD(p, rem, "Ntrip-Version: Ntrip/2.0\r\n");
        BUFADD(p, rem, "User-Agent: NTRIP %s\r\n", NTRIP_AGENT);
        if (*ntrip->user) {
            snprintf(user, sizeof(user), "%s:%s", ntrip->user, ntrip->passwd);
            BUFADD(p, rem, "Authorization: Basic ");
            if (rem > 0) {
                int b64_need = (int)(((strlen(user) + 2) / 3) * 4 + 1);
                if (b64_need > rem) {
                    sprintf(msg, "request buffer overflow");
                    return 0;
                }
                int n = encbase64(p, (uint8_t*)user, strlen(user));
                p += n;
                rem -= n;
            }
            BUFADD(p, rem, "\r\n");
        }
        BUFADD(p, rem, "Transfer-Encoding: chunked\r\n");
        BUFADD(p, rem, "\r\n");
    } else {
        /* NTRIP v1: SOURCE command */
        BUFADD(p, rem, "SOURCE %s %s\r\n", ntrip->passwd, ntrip->mntpnt);
        BUFADD(p, rem, "Source-Agent: NTRIP %s\r\n", NTRIP_AGENT);
        BUFADD(p, rem, "STR: %s\r\n", ntrip->str);
        BUFADD(p, rem, "\r\n");
    }
 
    if (rem <= 0) {
        sprintf(msg, "request buffer overflow");
        return 0;
    }
    if (writetcpcli(ntrip->tcp, (uint8_t*)buff, (int)(p - buff), msg) != (int)(p - buff)) {
        return 0;
    }
 
    tracet(NULL, 3, "reqntrip_s: send request state=%d ns=%d v2=%d\n", ntrip->state, (int)(p - buff), use_v2);
    tracet(NULL, 5, "reqntrip_s: n=%d buff=\n%s\n", (int)(p - buff), buff);
    ntrip->state = 1;
    return 1;
}
/* send ntrip client request -------------------------------------------------*/
static int reqntrip_c(ntrip_t* ntrip, char* msg) {
    char buff[2048], user[514], *p = buff;
    int rem = (int)sizeof(buff);
    int use_v2 = (ntrip->ver == NTRIP_VER_2 ||
                  (ntrip->ver == NTRIP_VER_AUTO && (!ntrip->v2_tried || ntrip->ver_neg == NTRIP_VER_2)));
 
    tracet(NULL, 3, "reqntrip_c: state=%d ver=%d\n", ntrip->state, ntrip->ver);
 
    if (use_v2) {
        /* NTRIP v2: HTTP/1.1 with Host and Ntrip-Version headers */
        BUFADD(p, rem, "GET %s/%s HTTP/1.1\r\n", ntrip->url, ntrip->mntpnt);
        BUFADD(p, rem, "Host: %s\r\n", ntrip->host);
        BUFADD(p, rem, "Ntrip-Version: Ntrip/2.0\r\n");
        BUFADD(p, rem, "User-Agent: NTRIP %s\r\n", NTRIP_AGENT);
 
        if (*ntrip->user) {
            snprintf(user, sizeof(user), "%s:%s", ntrip->user, ntrip->passwd);
            BUFADD(p, rem, "Authorization: Basic ");
            if (rem > 0) {
                int b64_need = (int)(((strlen(user) + 2) / 3) * 4 + 1);
                if (b64_need > rem) {
                    sprintf(msg, "request buffer overflow");
                    return 0;
                }
                int n = encbase64(p, (uint8_t*)user, strlen(user));
                p += n;
                rem -= n;
            }
            BUFADD(p, rem, "\r\n");
        }
        BUFADD(p, rem, "Connection: close\r\n");
        BUFADD(p, rem, "\r\n");
    } else {
        /* NTRIP v1: HTTP/1.0 (original) */
        BUFADD(p, rem, "GET %s/%s HTTP/1.0\r\n", ntrip->url, ntrip->mntpnt);
        BUFADD(p, rem, "User-Agent: NTRIP %s\r\n", NTRIP_AGENT);
 
        if (!*ntrip->user) {
            BUFADD(p, rem, "Accept: */*\r\n");
            BUFADD(p, rem, "Connection: close\r\n");
        } else {
            snprintf(user, sizeof(user), "%s:%s", ntrip->user, ntrip->passwd);
            BUFADD(p, rem, "Authorization: Basic ");
            if (rem > 0) {
                int b64_need = (int)(((strlen(user) + 2) / 3) * 4 + 1);
                if (b64_need > rem) {
                    sprintf(msg, "request buffer overflow");
                    return 0;
                }
                int n = encbase64(p, (uint8_t*)user, strlen(user));
                p += n;
                rem -= n;
            }
            BUFADD(p, rem, "\r\n");
        }
        BUFADD(p, rem, "\r\n");
    }
 
    if (rem <= 0) {
        sprintf(msg, "request buffer overflow");
        return 0;
    }
    if (writetcpcli(ntrip->tcp, (uint8_t*)buff, (int)(p - buff), msg) != (int)(p - buff)) {
        return 0;
    }
 
    tracet(NULL, 3, "reqntrip_c: send request state=%d ns=%d v2=%d\n", ntrip->state, (int)(p - buff), use_v2);
    tracet(NULL, 5, "reqntrip_c: n=%d buff=\n%s\n", (int)(p - buff), buff);
    ntrip->state = 1;
    return 1;
}
/* test ntrip server response ------------------------------------------------*/
static int rspntrip_s(ntrip_t* ntrip, char* msg) {
    int i, nb, body_off, status;
    char *p, *q;
 
    tracet(NULL, 3, "rspntrip_s: state=%d nb=%d\n", ntrip->state, ntrip->nb);
    ntrip->buff[ntrip->nb] = '\0';
    tracet(NULL, 5, "rspntrip_s: n=%d buff=\n%s\n", ntrip->nb, ntrip->buff);
 
    if ((p = strstr((char*)ntrip->buff, NTRIP_RSP_OK_SVR))) { /* v1 ok */
        q = (char*)ntrip->buff;
        p += strlen(NTRIP_RSP_OK_SVR);
        ntrip->nb -= (int)(p - q);
        for (i = 0; i < ntrip->nb; i++) {
            *q++ = *p++;
        }
        ntrip->state = 2;
        ntrip->ver_neg = NTRIP_VER_1;
        sprintf(msg, "%s/%s", ntrip->tcp->svr.saddr, ntrip->mntpnt);
        tracet(NULL, 3, "rspntrip_s: v1 response ok nb=%d\n", ntrip->nb);
        return 1;
    }
    /* check for HTTP/1.1 response (NTRIP v2 caster) */
    if ((body_off = http_header_end(ntrip->buff, ntrip->nb)) > 0) {
        status = http_status_code(ntrip->buff, ntrip->nb);
 
        if (status == 200) {
            ntrip->state = 2;
            ntrip->ver_neg = NTRIP_VER_2;
            /* shift remaining data to start of buffer */
            ntrip->nb -= body_off;
            if (ntrip->nb > 0) {
                memmove(ntrip->buff, ntrip->buff + body_off, ntrip->nb);
            }
            sprintf(msg, "%s/%s", ntrip->tcp->svr.saddr, ntrip->mntpnt);
            tracet(NULL, 3, "rspntrip_s: v2 response ok nb=%d\n", ntrip->nb);
            return 1;
        }
        /* HTTP error response */
        snprintf(msg, MAXSTATMSG, "%s", (char*)ntrip->buff);
        tracet(NULL, 3, "rspntrip_s: http error %s nb=%d\n", msg, ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
        return 0;
    }
    if ((p = strstr((char*)ntrip->buff, NTRIP_RSP_ERROR))) { /* v1 error */
        nb = ntrip->nb < MAXSTATMSG ? ntrip->nb : MAXSTATMSG;
        sprintf(msg, "%.*s", nb, (char*)ntrip->buff);
        if ((p = strchr(msg, '\r'))) {
            *p = '\0';
        }
        tracet(NULL, 3, "rspntrip_s: %s nb=%d\n", msg, ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
    } else if (ntrip->nb >= NTRIP_MAXRSP) { /* buffer overflow */
        sprintf(msg, "response overflow");
        tracet(NULL, 3, "rspntrip_s: response overflow nb=%d\n", ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
    }
    tracet(NULL, 5, "rspntrip_s: exit state=%d nb=%d\n", ntrip->state, ntrip->nb);
    return 0;
}
/* test ntrip client response ------------------------------------------------*/
static int rspntrip_c(ntrip_t* ntrip, char* msg) {
    int i, body_off, status;
    char *p, *q, val[256];
 
    tracet(NULL, 3, "rspntrip_c: state=%d nb=%d\n", ntrip->state, ntrip->nb);
    ntrip->buff[ntrip->nb] = '\0';
    tracet(NULL, 5, "rspntrip_c: n=%d buff=\n%s\n", ntrip->nb, ntrip->buff);
 
    if ((p = strstr((char*)ntrip->buff, NTRIP_RSP_OK_CLI))) { /* v1 ok: ICY 200 OK */
        q = (char*)ntrip->buff;
        p += strlen(NTRIP_RSP_OK_CLI);
        ntrip->nb -= (int)(p - q);
        for (i = 0; i < ntrip->nb; i++) {
            *q++ = *p++;
        }
        ntrip->state = 2;
        ntrip->ver_neg = NTRIP_VER_1;
        ntrip->chunked = 0;
        if (ntrip->ver == NTRIP_VER_AUTO) {
            ntrip->v2_tried = 1; /* accepted v2 request with v1 response */
        }
        sprintf(msg, "%s/%s", ntrip->tcp->svr.saddr, ntrip->mntpnt);
        tracet(NULL, 3, "rspntrip_c: v1 response ok nb=%d\n", ntrip->nb);
        return 1;
    }
    /* check for HTTP/1.1 response (NTRIP v2) */
    if ((body_off = http_header_end(ntrip->buff, ntrip->nb)) > 0) {
        status = http_status_code(ntrip->buff, ntrip->nb);
 
        if (status == 200) {
            /* check for NTRIP v2 source table */
            if (http_find_header(ntrip->buff, body_off, "Content-Type", val, sizeof(val)) &&
                strstr(val, "sourcetable")) {
                if (!*ntrip->mntpnt) { /* source table request */
                    int is_chunked = 0;
                    ntrip->state = 2;
                    ntrip->ver_neg = NTRIP_VER_2;
                    /* check for chunked encoding before shifting headers away */
                    if (http_find_header(ntrip->buff, body_off, "Transfer-Encoding", val, sizeof(val)) &&
                        stristr(val, "chunked")) {
                        is_chunked = 1;
                    }
                    /* shift body to start of buffer */
                    ntrip->nb -= body_off;
                    memmove(ntrip->buff, ntrip->buff + body_off, ntrip->nb);
                    /* enable persistent chunked decoding for incremental reads.
                     * ntrip->buff remains encoded; readntrip() decodes on demand */
                    if (is_chunked) {
                        ntrip->chunked = 1;
                        chunk_dec_init(&ntrip->cdec);
                    }
                    sprintf(msg, "source table received");
                    tracet(NULL, 3, "rspntrip_c: v2 source table nb=%d\n", ntrip->nb);
                    return 1;
                }
                sprintf(msg, "no mountp. reconnect...");
                tracet(NULL, 2, "rspntrip_c: v2 no mount point nb=%d\n", ntrip->nb);
                ntrip->nb = 0;
                ntrip->buff[0] = '\0';
                ntrip->state = 0;
                discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
                return 0;
            }
            /* NTRIP v2 data stream OK */
            ntrip->state = 2;
            ntrip->ver_neg = NTRIP_VER_2;
 
            /* detect chunked transfer encoding */
            if (http_find_header(ntrip->buff, body_off, "Transfer-Encoding", val, sizeof(val)) &&
                stristr(val, "chunked")) {
                ntrip->chunked = 1;
                chunk_dec_init(&ntrip->cdec);
            }
            /* shift remaining body data to start of buffer */
            ntrip->nb -= body_off;
            if (ntrip->nb > 0) {
                memmove(ntrip->buff, ntrip->buff + body_off, ntrip->nb);
            }
            sprintf(msg, "%s/%s", ntrip->tcp->svr.saddr, ntrip->mntpnt);
            tracet(NULL, 3, "rspntrip_c: v2 response ok chunked=%d nb=%d\n", ntrip->chunked, ntrip->nb);
            return 1;
        }
        /* HTTP error response */
        snprintf(msg, MAXSTATMSG, "%s", (char*)ntrip->buff);
        tracet(NULL, 3, "rspntrip_c: http error %s nb=%d\n", msg, ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
        return 0;
    }
    if ((p = strstr((char*)ntrip->buff, NTRIP_RSP_SRCTBL))) { /* v1 source table */
        if (!*ntrip->mntpnt) {                                /* source table request */
            ntrip->state = 2;
            ntrip->ver_neg = NTRIP_VER_1;
            sprintf(msg, "source table received");
            tracet(NULL, 3, "rspntrip_c: receive source table nb=%d\n", ntrip->nb);
            return 1;
        }
        sprintf(msg, "no mountp. reconnect...");
        tracet(NULL, 2, "rspntrip_c: no mount point nb=%d\n", ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
    } else if (ntrip->nb >= NTRIP_MAXRSP) { /* buffer overflow */
        sprintf(msg, "response overflow");
        tracet(NULL, 2, "rspntrip_c: response overflow nb=%d\n", ntrip->nb);
        ntrip->nb = 0;
        ntrip->buff[0] = '\0';
        ntrip->state = 0;
        discontcp(&ntrip->tcp->svr, ntrip->tcp->tirecon);
    }
    tracet(NULL, 5, "rspntrip_c: exit state=%d nb=%d\n", ntrip->state, ntrip->nb);
    return 0;
}
/* wait ntrip request/response -----------------------------------------------*/
static int waitntrip(ntrip_t* ntrip, char* msg) {
    int n;
    char* p;
 
    tracet(NULL, 4, "waitntrip: state=%d nb=%d\n", ntrip->state, ntrip->nb);
 
    if (ntrip->state < 0) {
        return 0; /* error */
    }
 
    if (ntrip->tcp->svr.state < 2) {
        ntrip->state = 0; /* tcp disconnected */
 
        /* AUTO fallback: only switch to v1 if connection dropped before any
         * response was received. If an HTTP response was seen (nb>0), the
         * caster supports v2 and the error is auth/mountpoint, not protocol */
        if (ntrip->ver == NTRIP_VER_AUTO && ntrip->v2_tried && ntrip->ver_neg == 0 && ntrip->nb == 0) {
            tracet(NULL, 3, "waitntrip: v2 failed before response, falling back to v1\n");
            ntrip->v2_tried = 2; /* mark as fallen back */
        }
    }
 
    if (ntrip->state == 0) { /* send request */
        int req_ok;
        if (ntrip->type == 0) {
            req_ok = reqntrip_s(ntrip, msg);
        } else {
            req_ok = reqntrip_c(ntrip, msg);
        }
        if (!req_ok) {
            return 0;
        }
        /* mark v2 as tried for AUTO mode */
        if (ntrip->ver == NTRIP_VER_AUTO && !ntrip->v2_tried) {
            ntrip->v2_tried = 1;
        }
        tracet(NULL, 3, "waitntrip: state=%d nb=%d\n", ntrip->state, ntrip->nb);
    }
    if (ntrip->state == 1) { /* read response */
        p = (char*)ntrip->buff + ntrip->nb;
        if ((n = readtcpcli(ntrip->tcp, (uint8_t*)p, NTRIP_MAXRSP - ntrip->nb - 1, msg)) == 0) {
            tracet(NULL, 5, "waitntrip: readtcp n=%d\n", n);
            return 0;
        }
        ntrip->nb += n;
        ntrip->buff[ntrip->nb] = '\0';
 
        /* wait response */
        if (ntrip->type == 0) {
            return rspntrip_s(ntrip, msg);
        } else {
            return rspntrip_c(ntrip, msg);
        }
    }
    return 1;
}
/* parse query parameters from mountpoint string
 * supported: ?ver=N (NTRIP version), &host=HOSTNAME (Host header override)
 * strips the query string from mntpnt
 * query string is copied to a local buffer to avoid modifying mntpnt beyond '?' */
static int parse_ntrip_query(char* mntpnt, char* host_override, int host_size) {
    char params[256], *p, *next;
    char* q;
    int ver = ntrip_ver_default;
 
    host_override[0] = '\0';
 
    q = strchr(mntpnt, '?');
    if (!q) {
        return ver;
    }
 
    /* copy query string to local buffer before truncating mntpnt */
    snprintf(params, sizeof(params), "%s", q + 1);
    *q = '\0'; /* strip query string from mountpoint */
 
    tracet(NULL, 3, "parse_ntrip_query: mntpnt=%s params=%s\n", mntpnt, params);
 
    /* parse &-separated parameters manually (no strtok) */
    p = params;
    while (*p) {
        /* find end of current parameter */
        next = strchr(p, '&');
        if (next) {
            *next = '\0';
        }
 
        if (strncmp(p, "ver=", 4) == 0) {
            int v = atoi(p + 4);
            if (v == 1) {
                ver = NTRIP_VER_1;
            } else if (v == 2) {
                ver = NTRIP_VER_2;
            }
        } else if (strncmp(p, "host=", 5) == 0) {
            const char* h = p + 5;
            int valid = 1, hlen = 0;
            /* reject control characters and whitespace to prevent header injection */
            while (h[hlen]) {
                if ((unsigned char)h[hlen] <= 0x20 || h[hlen] == 0x7F) {
                    valid = 0;
                    break;
                }
                hlen++;
            }
            if (valid && hlen > 0) {
                snprintf(host_override, host_size, "%s", h);
            }
        }
 
        if (next) {
            p = next + 1;
        } else {
            break;
        }
    }
    tracet(NULL, 3, "parse_ntrip_query: ver=%d host_override=%s\n", ver, host_override);
    return ver;
}
/* open ntrip ----------------------------------------------------------------*/
static ntrip_t* openntrip(const char* path, int type, char* msg) {
    ntrip_t* ntrip;
    int i;
    char addr[256] = "", port[256] = "", tpath[MAXSTRPATH];
 
    tracet(NULL, 3, "openntrip: path=%s type=%d\n", path, type);
 
    if (!(ntrip = (ntrip_t*)malloc(sizeof(ntrip_t)))) {
        return NULL;
    }
 
    ntrip->state = 0;
    ntrip->type = type; /* 0:server,1:client */
    ntrip->nb = 0;
    ntrip->ver = NTRIP_VER_AUTO;
    ntrip->ver_neg = 0;
    ntrip->chunked = 0;
    ntrip->v2_tried = 0;
    ntrip->url[0] = '\0';
    ntrip->host[0] = '\0';
    ntrip->mntpnt[0] = ntrip->user[0] = ntrip->passwd[0] = ntrip->str[0] = '\0';
    for (i = 0; i < NTRIP_MAXRSP; i++) {
        ntrip->buff[i] = 0;
    }
    chunk_dec_init(&ntrip->cdec);
 
    /* decode tcp/ntrip path */
    decodetcppath(path, addr, port, ntrip->user, ntrip->passwd, ntrip->mntpnt, ntrip->str);
 
    /* parse query parameters (?ver=N&host=HOSTNAME) from mountpoint */
    {
        char host_override[256] = "";
        ntrip->ver = parse_ntrip_query(ntrip->mntpnt, host_override, sizeof(host_override));
 
        /* use default port if no port specified */
        if (!*port) {
            sprintf(port, "%d", type ? NTRIP_CLI_PORT : NTRIP_SVR_PORT);
        }
        sprintf(tpath, "%s:%s", addr, port);
 
        /* save host header value (host= override takes priority) */
        if (*host_override) {
            snprintf(ntrip->host, sizeof(ntrip->host), "%s", host_override);
        } else {
            snprintf(ntrip->host, sizeof(ntrip->host), "%s:%s", addr, port);
        }
    }
 
    /* ntrip access via proxy server */
    if (*proxyaddr) {
        sprintf(ntrip->url, "http://%.*s", MAXSTRPATH - 8, tpath);
        sprintf(tpath, "%.*s", MAXSTRPATH - 1, proxyaddr);
    }
    /* open tcp client stream */
    if (!(ntrip->tcp = opentcpcli(tpath, msg))) {
        tracet(NULL, 2, "openntrip: opentcp error\n");
        free(ntrip);
        return NULL;
    }
    tracet(NULL, 3, "openntrip: ver=%d host=%s mntpnt=%s\n", ntrip->ver, ntrip->host, ntrip->mntpnt);
    return ntrip;
}
/* close ntrip ---------------------------------------------------------------*/
static void closentrip(ntrip_t* ntrip) {
    tracet(NULL, 3, "closentrip: state=%d\n", ntrip->state);
 
    /* send chunked terminator for v2 server before closing */
    if (ntrip->ver_neg == NTRIP_VER_2 && ntrip->type == 0 && ntrip->state == 2) {
        char msg_tmp[MAXSTRMSG] = "";
        uint8_t term[] = "0\r\n\r\n";
        writetcpcli(ntrip->tcp, term, 5, msg_tmp);
    }
    closetcpcli(ntrip->tcp);
    free(ntrip);
}
/* read ntrip ----------------------------------------------------------------*/
static int readntrip(ntrip_t* ntrip, uint8_t* buff, int n, char* msg) {
    int nb;
 
    tracet(NULL, 4, "readntrip:\n");
 
    if (!waitntrip(ntrip, msg)) {
        return 0;
    }
 
    if (ntrip->nb > 0) { /* read response buffer first */
        if (ntrip->chunked) {
            /* decode directly from ntrip->buff into caller's buff */
            const uint8_t* in = ntrip->buff;
            int nin = ntrip->nb;
            int nd = chunk_decode(&ntrip->cdec, &in, &nin, buff, n);
            if (nd < 0) {
                tracet(NULL, 2, "readntrip: chunk decode error, resetting\n");
                ntrip->nb = 0;
                chunk_dec_init(&ntrip->cdec);
                return 0;
            }
            /* compact: remove consumed bytes from ntrip->buff */
            if (nin > 0) {
                memmove(ntrip->buff, in, nin);
            }
            ntrip->nb = nin;
            return nd;
        }
        nb = ntrip->nb <= n ? ntrip->nb : n;
        memcpy(buff, ntrip->buff, nb);
        if (nb < ntrip->nb) {
            memmove(ntrip->buff, ntrip->buff + nb, ntrip->nb - nb);
        }
        ntrip->nb -= nb;
        return nb;
    }
    if (ntrip->chunked) {
        /* read raw TCP data and decode chunks */
        uint8_t raw[4096];
        int nr = readtcpcli(ntrip->tcp, raw, sizeof(raw), msg);
        if (nr <= 0) {
            return 0;
        }
        const uint8_t* in = raw;
        int nin = nr;
        int nd = chunk_decode(&ntrip->cdec, &in, &nin, buff, n);
        if (nd < 0) {
            tracet(NULL, 2, "readntrip: chunk decode error, resetting\n");
            ntrip->nb = 0;
            chunk_dec_init(&ntrip->cdec);
            return 0;
        }
        /* preserve unconsumed encoded bytes to response buffer */
        if (nin > 0) {
            if (ntrip->nb + nin <= NTRIP_MAXRSP) {
                memcpy(ntrip->buff + ntrip->nb, in, nin);
                ntrip->nb += nin;
            } else {
                tracet(NULL, 2, "readntrip: chunk buffer overflow, resetting\n");
                ntrip->nb = 0;
                chunk_dec_init(&ntrip->cdec);
            }
        }
        return nd;
    }
    return readtcpcli(ntrip->tcp, buff, n, msg);
}
/* write ntrip ---------------------------------------------------------------*/
static int writentrip(ntrip_t* ntrip, uint8_t* buff, int n, char* msg) {
    tracet(NULL, 3, "writentrip: n=%d\n", n);
 
    if (!waitntrip(ntrip, msg)) {
        return 0;
    }
 
    /* v2 server: use chunked transfer encoding */
    if (ntrip->ver_neg == NTRIP_VER_2 && ntrip->type == 0) {
        uint8_t cbuf[4096 + 20];
        int remaining = n, sent = 0;
 
        while (remaining > 0) {
            int chunk = remaining < 4096 ? remaining : 4096;
            int cn = chunk_encode(cbuf, sizeof(cbuf), buff + sent, chunk);
            if (cn < 0 || writetcpcli(ntrip->tcp, cbuf, cn, msg) != cn) {
                return sent;
            }
            sent += chunk;
            remaining -= chunk;
        }
        return sent;
    }
    return writetcpcli(ntrip->tcp, buff, n, msg);
}
/* get state ntrip -----------------------------------------------------------*/
static int statentrip(ntrip_t* ntrip) {
    return !ntrip ? 0 : (ntrip->state == 0 ? ntrip->tcp->svr.state : ntrip->state);
}
/* get extended state ntrip --------------------------------------------------*/
static int statexntrip(ntrip_t* ntrip, char* msg) {
    char* p = msg;
    int state = !ntrip ? 0 : (ntrip->state == 0 ? ntrip->tcp->svr.state : ntrip->state);
 
    p += sprintf(p, "ntrip:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  state   = %d\n", state);
    p += sprintf(p, "  type    = %d\n", ntrip->type);
    p += sprintf(p, "  ver     = %d\n", ntrip->ver);
    p += sprintf(p, "  ver_neg = %d\n", ntrip->ver_neg);
    p += sprintf(p, "  chunked = %d\n", ntrip->chunked);
    p += sprintf(p, "  nb      = %d\n", ntrip->nb);
    p += sprintf(p, "  url     = %s\n", ntrip->url);
    p += sprintf(p, "  host    = %s\n", ntrip->host);
    p += sprintf(p, "  mntpnt  = %s\n", ntrip->mntpnt);
    p += sprintf(p, "  user    = %s\n", ntrip->user);
    p += sprintf(p, "  passwd  = %s\n", ntrip->passwd);
    p += sprintf(p, "  str     = %s\n", ntrip->str);
    p += sprintf(p, "  svr:\n");
    p += statextcp(&ntrip->tcp->svr, p);
    return state;
}
/* open ntrip-caster ---------------------------------------------------------*/
static ntripc_t* openntripc(const char* path, char* msg) {
    ntripc_t* ntripc;
    int i;
    char port[256] = "", tpath[MAXSTRPATH];
 
    tracet(NULL, 3, "openntripc: path=%s\n", path);
 
    if (!(ntripc = (ntripc_t*)malloc(sizeof(ntripc_t)))) {
        return NULL;
    }
 
    ntripc->state = 0;
    ntripc->mntpnt[0] = ntripc->user[0] = ntripc->passwd[0] = ntripc->srctbl[0] = '\0';
    for (i = 0; i < MAXCLI; i++) {
        ntripc->con[i].state = 0;
        ntripc->con[i].ver = 1;
        ntripc->con[i].nb = 0;
        memset(ntripc->con[i].buff, 0, NTRIP_MAXRSP);
    }
    /* decode tcp/ntrip path */
    decodetcppath(path, NULL, port, ntripc->user, ntripc->passwd, ntripc->mntpnt, ntripc->srctbl);
 
    if (!*ntripc->mntpnt) {
        tracet(NULL, 2, "openntripc: no mountpoint path=%s\n", path);
        free(ntripc);
        return NULL;
    }
    /* use default port if no port specified */
    if (!*port) {
        sprintf(port, "%d", NTRIP_CLI_PORT);
    }
    sprintf(tpath, ":%s", port);
 
    /* open tcp server stream */
    if (!(ntripc->tcp = opentcpsvr(tpath, msg))) {
        tracet(NULL, 2, "openntripc: opentcpsvr error port=%d\n", port);
        free(ntripc);
        return NULL;
    }
    return ntripc;
}
/* close ntrip-caster --------------------------------------------------------*/
static void closentripc(ntripc_t* ntripc) {
    int i;
    tracet(NULL, 3, "closentripc: state=%d\n", ntripc->state);
 
    /* send chunked terminator for all connected v2 clients */
    for (i = 0; i < MAXCLI; i++) {
        if (ntripc->con[i].state && ntripc->con[i].ver >= 2) {
            uint8_t term[] = "0\r\n\r\n";
            send_nb(ntripc->tcp->cli[i].sock, term, 5);
        }
    }
    closetcpsvr(ntripc->tcp);
    free(ntripc);
}
/* disconnect ntrip-caster connection ----------------------------------------*/
static void discon_ntripc(ntripc_t* ntripc, int i) {
    tracet(NULL, 3, "discon_ntripc: i=%d\n", i);
 
    /* send chunked terminator for v2 clients before disconnecting */
    if (ntripc->con[i].ver >= 2 && ntripc->con[i].state) {
        uint8_t term[] = "0\r\n\r\n";
        send_nb(ntripc->tcp->cli[i].sock, term, 5);
    }
    discontcp(&ntripc->tcp->cli[i], ticonnect);
    ntripc->con[i].nb = 0;
    ntripc->con[i].buff[0] = '\0';
    ntripc->con[i].ver = 1;
    ntripc->con[i].state = 0;
}
/* http date-time string -----------------------------------------------------*/
static void time_str_http(char* buff) {
    static const char* month[] = {"Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"};
    static const char* wkday[] = {"Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"};
    time_t tt = time(NULL);
    struct tm* tm = gmtime(&tt);
 
    sprintf(buff, "%s, %02d %s %04d %02d:%02d:%02d GMT", wkday[tm->tm_wday], tm->tm_mday, month[tm->tm_mon],
            1900 + tm->tm_year, tm->tm_hour, tm->tm_min, tm->tm_sec);
}
/* send NTRIP source table ---------------------------------------------------*/
static void send_srctbl(ntripc_t* ntripc, int ver, socket_t sock) {
    char srctbl[512 + NTRIP_MAXSTR], buff[256], tstr[64], *p = buff;
 
    sprintf(srctbl, "STR;%s;%s\r\n%s\r\n", ntripc->mntpnt, ntripc->srctbl, NTRIP_RSP_TBLEND);
 
    if (ver == 1) { /* NTRIP ver.1 */
        p += sprintf(p, "%s", NTRIP_RSP_SRCTBL);
    } else { /* NTRIP ver.2 */
        p += sprintf(p, "HTTP/1.1 200 OK\r\n");
        p += sprintf(p, "Ntrip-Version: Ntrip/2.0\r\n");
        p += sprintf(p, "Ntrip-Flags: \r\n");
    }
    time_str_http(tstr);
    p += sprintf(p, "Server: NTRIP %s/%s\r\n", MRTKLIB_SOFTNAME, MRTKLIB_VERSION_STRING);
    p += sprintf(p, "Date: %s\r\n", tstr);
    p += sprintf(p, "Connection: close\r\n");
    p += sprintf(p, "Content-Type: %s\r\n", (ver == 1) ? "text/plain" : "gnss/sourcetable");
    p += sprintf(p, "Content-Length: %d\r\n\r\n", (int)strlen(srctbl));
    send_nb(sock, (uint8_t*)buff, (int)(p - buff));
    send_nb(sock, (uint8_t*)srctbl, (int)strlen(srctbl));
}
/* send OK response ----------------------------------------------------------*/
static void send_rsp_ok(int ver, socket_t sock) {
    char buff[512], tstr[64], *p = buff;
 
    if (ver == 1) { /* NTRIP ver.1 */
        p += sprintf(p, "%s", NTRIP_RSP_OK_CLI);
    } else { /* NTRIP ver.2 */
        time_str_http(tstr);
        p += sprintf(p, "HTTP/1.1 200 OK\r\n");
        p += sprintf(p, "Ntrip-Version: Ntrip/2.0\r\n");
        p += sprintf(p, "Server: NTRIP %s/%s\r\n", MRTKLIB_SOFTNAME, MRTKLIB_VERSION_STRING);
        p += sprintf(p, "Date: %s\r\n", tstr);
        p += sprintf(p, "Cache-Control: no-store, no-cache, max-age=0\r\n");
        p += sprintf(p, "Pragma: no-cache\r\n");
        p += sprintf(p, "Transfer-Encoding: chunked\r\n");
        p += sprintf(p, "Connection: close\r\n");
        p += sprintf(p, "Content-Type: gnss/data\r\n\r\n");
    }
    send_nb(sock, (uint8_t*)buff, (int)(p - buff));
}
/* send error response -------------------------------------------------------*/
static void send_rsp_err(ntripc_t* ntripc, int ver, int err, socket_t sock) {
    char buff[512], tstr[64], *p = buff;
 
    if (ver == 1) { /* NTRIP ver.1 */
        if (err == 404) {
            send_srctbl(ntripc, ver, sock); /* send source table if no mp */
            return;
        }
        p += sprintf(p, "%s", NTRIP_RSP_UNAUTH);
    } else { /* NTRIP ver.2 */
        time_str_http(tstr);
        p += sprintf(p, "HTTP/1.1 %d %s\r\n", err, (err == 404) ? "Not Found" : "Unauthorized");
        p += sprintf(p, "Ntrip-Version: Ntrip/2.0\r\n");
        p += sprintf(p, "Server: NTRIP %s/%s\r\n", MRTKLIB_SOFTNAME, MRTKLIB_VERSION_STRING);
        p += sprintf(p, "Date: %s\r\n", tstr);
        p += sprintf(p, "Content-Type: text/html\r\n");
        p += sprintf(p, "Connection: close\r\n\r\n");
        p += sprintf(p, "%s\r\n", (err == 404) ? "Mountpoint of request not found" : "No or wrong authorization");
    }
    send_nb(sock, (uint8_t*)buff, (int)(p - buff));
}
/* test ntrip client request -------------------------------------------------*/
static void rsp_ntripc(ntripc_t* ntripc, int i) {
    ntripc_con_t* con = ntripc->con + i;
    char url[256] = "", mntpnt[256] = "", proto[256] = "", user[513], user_pwd[256], *p, *q;
    int ver = 1;
 
    tracet(NULL, 3, "rsp_ntripc=%d\n", i);
    con->buff[con->nb] = '\0';
    tracet(NULL, 5, "rsp_ntripc: n=%d,buff=\n%s\n", con->nb, con->buff);
 
    if (con->nb >= NTRIP_MAXRSP - 1) { /* buffer overflow */
        tracet(NULL, 2, "rsp_ntripc: request buffer overflow\n");
        discon_ntripc(ntripc, i);
        return;
    }
    /* test GET and User-Agent */
    if (!(p = strstr((char*)con->buff, "GET")) || !(q = strstr(p, "\r\n")) || !(q = strstr(q, "User-Agent:")) ||
        !strstr(q, "\r\n")) {
        tracet(NULL, 2, "rsp_ntripc: NTRIP request error\n");
        discon_ntripc(ntripc, i);
        return;
    }
    /* test protocol */
    if (sscanf(p, "GET %255s %255s", url, proto) < 2 || (strcmp(proto, "HTTP/1.0") && strcmp(proto, "HTTP/1.1"))) {
        tracet(NULL, 2, "rsp_ntripc: NTRIP request error proto=%s\n", proto);
        discon_ntripc(ntripc, i);
        return;
    }
    /* test NTRIP version (clamp to 1 or 2) */
    if ((p = strstr((char*)con->buff, "Ntrip-Version:"))) {
        sscanf(p + 14, " Ntrip/%d", &ver);
        if (ver < 1 || ver > 2) {
            ver = 1; /* treat unsupported versions as v1 */
        }
    }
    if ((p = strchr(url, '/'))) {
        strcpy(mntpnt, p + 1);
    }
 
    /* source table request */
    if (!*mntpnt) {
        send_srctbl(ntripc, ver, ntripc->tcp->cli[i].sock);
        discon_ntripc(ntripc, i);
        return;
    }
    /* test mountpoint */
    if (strcmp(mntpnt, ntripc->mntpnt)) {
        send_rsp_err(ntripc, ver, 404, ntripc->tcp->cli[i].sock);
        discon_ntripc(ntripc, i);
        tracet(NULL, 2, "rsp_ntripc: no mountpoint %s\n", mntpnt);
        return;
    }
    /* test authentication */
    if (*ntripc->passwd) {
        sprintf(user, "%s:%s", ntripc->user, ntripc->passwd);
        q = user_pwd;
        q += sprintf(q, "Authorization: Basic ");
        q += encbase64(q, (uint8_t*)user, strlen(user));
        if (!(p = strstr((char*)con->buff, "Authorization:")) || strncmp(p, user_pwd, strlen(user_pwd))) {
            send_rsp_err(ntripc, ver, 401, ntripc->tcp->cli[i].sock);
            discon_ntripc(ntripc, i);
            tracet(NULL, 2, "rsp_ntripc: authorization error\n");
            return;
        }
    }
    /* send OK response */
    send_rsp_ok(ver, ntripc->tcp->cli[i].sock);
 
    con->state = 1;
    con->ver = ver;
    strcpy(con->mntpnt, mntpnt);
}
/* handle ntrip client connect request ---------------------------------------*/
static void wait_ntripc(ntripc_t* ntripc, char* msg) {
    uint8_t* buff;
    int i, n, nmax, err;
 
    tracet(NULL, 4, "wait_ntripc\n");
 
    ntripc->state = ntripc->tcp->svr.state;
 
    if (!waittcpsvr(ntripc->tcp, msg)) {
        return;
    }
 
    for (i = 0; i < MAXCLI; i++) {
        if (ntripc->tcp->cli[i].state != 2 || ntripc->con[i].state) {
            continue;
        }
 
        /* receive ntrip client request */
        buff = ntripc->con[i].buff + ntripc->con[i].nb;
        nmax = NTRIP_MAXRSP - ntripc->con[i].nb - 1;
 
        if ((n = recv_nb(ntripc->tcp->cli[i].sock, buff, nmax)) == -1) {
            if ((err = errsock())) {
                tracet(NULL, 2, "wait_ntripc: recv error sock=%d err=%d\n", ntripc->tcp->cli[i].sock, err);
            }
            discon_ntripc(ntripc, i);
            continue;
        }
        if (n <= 0) {
            continue;
        }
 
        /* test ntrip client request */
        ntripc->con[i].nb += n;
        rsp_ntripc(ntripc, i);
    }
}
/* read ntrip-caster ---------------------------------------------------------*/
static int readntripc(ntripc_t* ntripc, uint8_t* buff, int n, char* msg) {
    int i, nr, err;
 
    tracet(NULL, 4, "readntripc:\n");
 
    wait_ntripc(ntripc, msg);
 
    for (i = 0; i < MAXCLI; i++) {
        if (!ntripc->con[i].state) {
            continue;
        }
 
        nr = recv_nb(ntripc->tcp->cli[i].sock, buff, n);
 
        if (nr < 0) {
            if ((err = errsock())) {
                tracet(NULL, 2, "readntripc: recv error i=%d sock=%d err=%d\n", i, ntripc->tcp->cli[i].sock, err);
            }
            discon_ntripc(ntripc, i);
        } else if (nr > 0) {
            ntripc->tcp->cli[i].tact = tickget();
            return nr;
        }
    }
    return 0;
}
/* write ntrip-caster --------------------------------------------------------*/
static int writentripc(ntripc_t* ntripc, uint8_t* buff, int n, char* msg) {
    int i, ns = 0, err;
 
    tracet(NULL, 4, "writentripc: n=%d\n", n);
 
    wait_ntripc(ntripc, msg);
 
    for (i = 0; i < MAXCLI; i++) {
        if (!ntripc->con[i].state) {
            continue;
        }
 
        if (ntripc->con[i].ver >= 2) {
            /* v2 client: use chunked transfer encoding */
            uint8_t cbuf[4096 + 20];
            int remaining = n, sent = 0;
 
            while (remaining > 0) {
                int chunk = remaining < 4096 ? remaining : 4096;
                int cn = chunk_encode(cbuf, sizeof(cbuf), buff + sent, chunk);
                if (cn < 0) {
                    break;
                }
                if (send_nb(ntripc->tcp->cli[i].sock, cbuf, cn) < cn) {
                    if ((err = errsock())) {
                        tracet(NULL, 2, "writentripc: send error i=%d err=%d\n", i, err);
                    }
                    discon_ntripc(ntripc, i);
                    break;
                }
                sent += chunk;
                remaining -= chunk;
            }
            if (remaining <= 0) {
                ntripc->tcp->cli[i].tact = tickget();
                ns = n; /* return payload bytes, not encoded bytes */
            }
        } else {
            /* v1 client: raw data */
            ns = send_nb(ntripc->tcp->cli[i].sock, buff, n);
 
            if (ns < n) {
                if ((err = errsock())) {
                    tracet(NULL, 2, "writentripc: send error i=%d sock=%d err=%d\n", i, ntripc->tcp->cli[i].sock, err);
                }
                discon_ntripc(ntripc, i);
            } else {
                ntripc->tcp->cli[i].tact = tickget();
            }
        }
    }
    return ns;
}
/* get state ntrip-caster ----------------------------------------------------*/
static int statentripc(ntripc_t* ntripc) { return !ntripc ? 0 : ntripc->state; }
/* get extended state ntrip-caster -------------------------------------------*/
static int statexntripc(ntripc_t* ntripc, char* msg) {
    char* p = msg;
    int i, state = !ntripc ? 0 : ntripc->state;
 
    p += sprintf(p, "ntripc:\n");
    p += sprintf(p, "  state   = %d\n", ntripc->state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  type    = %d\n", ntripc->type);
    p += sprintf(p, "  mntpnt  = %s\n", ntripc->mntpnt);
    p += sprintf(p, "  user    = %s\n", ntripc->user);
    p += sprintf(p, "  passwd  = %s\n", ntripc->passwd);
    p += sprintf(p, "  srctbl  = %s\n", ntripc->srctbl);
    p += sprintf(p, "  svr:\n");
    p += statextcp(&ntripc->tcp->svr, p);
    for (i = 0; i < MAXCLI; i++) {
        if (!ntripc->tcp->cli[i].state) {
            continue;
        }
        p += sprintf(p, "  cli#%d:\n", i);
        p += statextcp(ntripc->tcp->cli + i, p);
        p += sprintf(p, "    mntpnt= %s\n", ntripc->con[i].mntpnt);
        p += sprintf(p, "    nb    = %d\n", ntripc->con[i].nb);
    }
    return state;
}
/* generate udp socket -------------------------------------------------------*/
static udp_t* genudp(int type, int port, const char* saddr, char* msg) {
    udp_t* udp;
    struct hostent* hp;
    int bs = buffsize, opt = 1;
 
    tracet(NULL, 3, "genudp: type=%d\n", type);
 
    if (!(udp = (udp_t*)malloc(sizeof(udp_t)))) {
        return NULL;
    }
    udp->state = 2;
    udp->type = type;
    udp->port = port;
    strcpy(udp->saddr, saddr);
 
    if ((udp->sock = socket(AF_INET, SOCK_DGRAM, 0)) == (socket_t)-1) {
        sprintf(msg, "socket error (%d)", errsock());
        return NULL;
    }
    if (setsockopt(udp->sock, SOL_SOCKET, SO_RCVBUF, (const char*)&bs, sizeof(bs)) == -1 ||
        setsockopt(udp->sock, SOL_SOCKET, SO_SNDBUF, (const char*)&bs, sizeof(bs)) == -1) {
        tracet(NULL, 2, "genudp: setsockopt error sock=%d err=%d bs=%d\n", udp->sock, errsock(), bs);
        sprintf(msg, "sockopt error: bufsiz");
    }
    memset(&udp->addr, 0, sizeof(udp->addr));
    udp->addr.sin_family = AF_INET;
    udp->addr.sin_port = htons(port);
 
    if (!udp->type) { /* udp server */
        udp->addr.sin_addr.s_addr = htonl(INADDR_ANY);
#ifdef SVR_REUSEADDR
        setsockopt(udp->sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt));
#endif
        if (bind(udp->sock, (struct sockaddr*)&udp->addr, sizeof(udp->addr)) == -1) {
            tracet(NULL, 2, "genudp: bind error sock=%d port=%d err=%d\n", udp->sock, port, errsock());
            sprintf(msg, "bind error (%d): %d", errsock(), port);
            closesocket(udp->sock);
            free(udp);
            return NULL;
        }
    } else { /* udp client */
        if (!strcmp(saddr, "255.255.255.255") &&
            setsockopt(udp->sock, SOL_SOCKET, SO_BROADCAST, (const char*)&opt, sizeof(opt)) == -1) {
            tracet(NULL, 2, "genudp: setsockopt error sock=%d err=%d\n", udp->sock, errsock());
            sprintf(msg, "sockopt error: broadcast");
        }
        if (!(hp = gethostbyname(saddr))) {
            sprintf(msg, "address error (%s)", saddr);
            closesocket(udp->sock);
            free(udp);
            return NULL;
        }
        memcpy(&udp->addr.sin_addr, hp->h_addr_list[0], hp->h_length);
    }
    return udp;
}
/* open udp server -----------------------------------------------------------*/
static udp_t* openudpsvr(const char* path, char* msg) {
    char sport[256] = "";
    int port;
 
    tracet(NULL, 3, "openudpsvr: path=%s\n", path);
 
    decodetcppath(path, NULL, sport, NULL, NULL, NULL, NULL);
 
    if (sscanf(sport, "%d", &port) < 1) {
        sprintf(msg, "port error: %s", sport);
        tracet(NULL, 2, "openudpsvr: port error port=%s\n", port);
        return NULL;
    }
    return genudp(0, port, "", msg);
}
/* close udp server ----------------------------------------------------------*/
static void closeudpsvr(udp_t* udpsvr) {
    tracet(NULL, 3, "closeudpsvr: sock=%d\n", udpsvr->sock);
 
    closesocket(udpsvr->sock);
    free(udpsvr);
}
/* read udp server -----------------------------------------------------------*/
static int readudpsvr(udp_t* udpsvr, uint8_t* buff, int n, char* msg) {
    struct timeval tv = {0};
    fd_set rs;
    int ret, nr;
 
    tracet(NULL, 4, "readudpsvr: sock=%d n=%d\n", udpsvr->sock, n);
 
    FD_ZERO(&rs);
    FD_SET(udpsvr->sock, &rs);
    ret = select(udpsvr->sock + 1, &rs, NULL, NULL, &tv);
    if (ret <= 0) {
        return ret;
    }
    nr = recvfrom(udpsvr->sock, (char*)buff, n, 0, NULL, NULL);
    return nr <= 0 ? -1 : nr;
}
/* get state udp server ------------------------------------------------------*/
static int stateudpsvr(udp_t* udpsvr) { return udpsvr ? udpsvr->state : 0; }
/* get extended state udp server ---------------------------------------------*/
static int statexudpsvr(udp_t* udpsvr, char* msg) {
    char* p = msg;
    int state = udpsvr ? udpsvr->state : 0;
 
    p += sprintf(p, "udpsvr:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  type    = %d\n", udpsvr->type);
    p += sprintf(p, "  sock    = %d\n", (int)udpsvr->sock);
    p += sprintf(p, "  port    = %d\n", udpsvr->port);
    return state;
}
/* open udp client -----------------------------------------------------------*/
static udp_t* openudpcli(const char* path, char* msg) {
    char sport[256] = "", saddr[256] = "";
    int port;
 
    tracet(NULL, 3, "openudpsvr: path=%s\n", path);
 
    decodetcppath(path, saddr, sport, NULL, NULL, NULL, NULL);
 
    if (sscanf(sport, "%d", &port) < 1) {
        sprintf(msg, "port error: %s", sport);
        tracet(NULL, 2, "openudpcli: port error port=%s\n", sport);
        return NULL;
    }
    return genudp(1, port, saddr, msg);
}
/* close udp client ----------------------------------------------------------*/
static void closeudpcli(udp_t* udpcli) {
    tracet(NULL, 3, "closeudpcli: sock=%d\n", udpcli->sock);
 
    closesocket(udpcli->sock);
    free(udpcli);
}
/* write udp client -----------------------------------------------------------*/
static int writeudpcli(udp_t* udpcli, uint8_t* buff, int n, char* msg) {
    tracet(NULL, 4, "writeudpcli: sock=%d n=%d\n", udpcli->sock, n);
 
    return (int)sendto(udpcli->sock, (char*)buff, n, 0, (struct sockaddr*)&udpcli->addr, sizeof(udpcli->addr));
}
/* get state udp client ------------------------------------------------------*/
static int stateudpcli(udp_t* udpcli) { return udpcli ? udpcli->state : 0; }
/* get extended state udp client ---------------------------------------------*/
static int statexudpcli(udp_t* udpcli, char* msg) {
    char* p = msg;
    int state = udpcli ? udpcli->state : 0;
 
    p += sprintf(p, "udpsvr:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  type    = %d\n", udpcli->type);
    p += sprintf(p, "  sock    = %d\n", (int)udpcli->sock);
    p += sprintf(p, "  addr    = %s\n", udpcli->saddr);
    p += sprintf(p, "  port    = %d\n", udpcli->port);
    return state;
}
/* decode ftp path -----------------------------------------------------------*/
static void decodeftppath(const char* path, char* addr, char* file, char* user, char* passwd, int* topts) {
    char buff[MAXSTRPATH], *p, *q;
 
    tracet(NULL, 4, "decodeftpath: path=%s\n", path);
 
    if (user) {
        *user = '\0';
    }
    if (passwd) {
        *passwd = '\0';
    }
    if (topts) {
        topts[0] = 0;    /* time offset in path (s) */
        topts[1] = 3600; /* download interval (s) */
        topts[2] = 0;    /* download time offset (s) */
        topts[3] = 0;    /* retry interval (s) (0: no retry) */
    }
    strcpy(buff, path);
 
    if ((p = strchr(buff, '/'))) {
        if ((q = strstr(p + 1, "::"))) {
            *q = '\0';
            if (topts) {
                sscanf(q + 2, "T=%d,%d,%d,%d", topts, topts + 1, topts + 2, topts + 3);
            }
        }
        strcpy(file, p + 1);
        *p = '\0';
    } else {
        file[0] = '\0';
    }
 
    if ((p = strrchr(buff, '@'))) {
        *p++ = '\0';
        if ((q = strchr(buff, ':'))) {
            *q = '\0';
            if (passwd) {
                strcpy(passwd, q + 1);
            }
        }
        *q = '\0';
        if (user) {
            strcpy(user, buff);
        }
    } else {
        p = buff;
    }
 
    strcpy(addr, p);
}
/* next download time --------------------------------------------------------*/
static gtime_t nextdltime(const int* topts, int stat) {
    gtime_t time;
    double tow;
    int week, tint;
 
    tracet(NULL, 3, "nextdltime: topts=%d %d %d %d stat=%d\n", topts[0], topts[1], topts[2], topts[3], stat);
 
    /* current time (gpst) */
    time = utc2gpst(timeget());
    tow = time2gpst(time, &week);
 
    /* next retry time */
    if (stat == 0 && topts[3] > 0) {
        tow = (floor((tow - topts[2]) / topts[3]) + 1.0) * topts[3] + topts[2];
        return gpst2time(week, tow);
    }
    /* next interval time */
    tint = topts[1] <= 0 ? 3600 : topts[1];
    tow = (floor((tow - topts[2]) / tint) + 1.0) * tint + topts[2];
    time = gpst2time(week, tow);
 
    return time;
}
/* ftp thread ----------------------------------------------------------------*/
static void* ftpthread(void* arg) {
    ftp_t* ftp = (ftp_t*)arg;
    FILE* fp;
    gtime_t time;
    char remote[1024], local[1024], tmpfile[1024], errfile[1024], *p;
    char cmd[5120], env[1024] = "", opt[1024], *proxyopt = "", *proto;
    int ret;
 
    tracet(NULL, 3, "ftpthread:\n");
 
    if (!*localdir) {
        tracet(NULL, 2, "no local directory\n");
        ftp->error = 11;
        ftp->state = 3;
        return 0;
    }
    /* replace keyword in file path and local path */
    time = timeadd(utc2gpst(timeget()), ftp->topts[0]);
    reppath(ftp->file, remote, time, "", "");
 
    if ((p = strrchr(remote, '/'))) {
        p++;
    } else {
        p = remote;
    }
    sprintf(local, "%.768s%c%.254s", localdir, FILEPATHSEP, p);
    sprintf(errfile, "%.1019s.err", local);
 
    /* if local file exist, skip download */
    strcpy(tmpfile, local);
    if ((p = strrchr(tmpfile, '.')) && (!strcmp(p, ".z") || !strcmp(p, ".gz") || !strcmp(p, ".zip") ||
                                        !strcmp(p, ".Z") || !strcmp(p, ".GZ") || !strcmp(p, ".ZIP"))) {
        *p = '\0';
    }
    if ((fp = fopen(tmpfile, "rb"))) {
        fclose(fp);
        sprintf(ftp->local, "%.1023s", tmpfile);
        tracet(NULL, 3, "ftpthread: file exists %s\n", ftp->local);
        ftp->state = 2;
        return 0;
    }
    /* proxy settings for wget (ref [2]) */
    if (*proxyaddr) {
        proto = ftp->proto ? "http" : "ftp";
        sprintf(env, "set %.4s_proxy=http://%.998s & ", proto, proxyaddr);
        proxyopt = "--proxy=on ";
    }
    /* download command (ref [2]) */
    if (ftp->proto == 0) { /* ftp */
        sprintf(opt,
                "--ftp-user=%.32s --ftp-password=%.32s --glob=off "
                "--passive-ftp %.32s -t 1 -T %d -O \"%.768s\"",
                ftp->user, ftp->passwd, proxyopt, FTP_TIMEOUT, local);
        sprintf(cmd, "%s%s %s \"ftp://%s/%s\" 2> \"%.768s\"\n", env, FTP_CMD, opt, ftp->addr, remote, errfile);
    } else { /* http */
        sprintf(opt, "%.32s -t 1 -T %d -O \"%.768s\"", proxyopt, FTP_TIMEOUT, local);
        sprintf(cmd, "%s%s %s \"http://%s/%s\" 2> \"%.768s\"\n", env, FTP_CMD, opt, ftp->addr, remote, errfile);
    }
    /* execute download command */
    if ((ret = execcmd(cmd))) {
        remove(local);
        tracet(NULL, 2, "execcmd error: cmd=%s ret=%d\n", cmd, ret);
        ftp->error = ret;
        ftp->state = 3;
        return 0;
    }
    remove(errfile);
 
    /* uncompress downloaded file */
    if ((p = strrchr(local, '.')) && (!strcmp(p, ".z") || !strcmp(p, ".gz") || !strcmp(p, ".zip") || !strcmp(p, ".Z") ||
                                      !strcmp(p, ".GZ") || !strcmp(p, ".ZIP"))) {
        if (rtk_uncompress(local, tmpfile)) {
            remove(local);
            strcpy(local, tmpfile);
        } else {
            tracet(NULL, 2, "file uncompact error: %s\n", local);
            ftp->error = 12;
            ftp->state = 3;
            return 0;
        }
    }
    strcpy(ftp->local, local);
    ftp->state = 2; /* ftp completed */
 
    tracet(NULL, 3, "ftpthread: complete cmd=%s\n", cmd);
    return 0;
}
/* open ftp ------------------------------------------------------------------*/
static ftp_t* openftp(const char* path, int type, char* msg) {
    ftp_t* ftp;
 
    tracet(NULL, 3, "openftp: path=%s type=%d\n", path, type);
 
    msg[0] = '\0';
 
    if (!(ftp = (ftp_t*)malloc(sizeof(ftp_t)))) {
        return NULL;
    }
 
    ftp->state = 0;
    ftp->proto = type;
    ftp->error = 0;
    ftp->thread = 0;
    ftp->local[0] = '\0';
 
    /* decode ftp path */
    decodeftppath(path, ftp->addr, ftp->file, ftp->user, ftp->passwd, ftp->topts);
 
    /* set first download time */
    ftp->tnext = timeadd(timeget(), 10.0);
 
    return ftp;
}
/* close ftp -----------------------------------------------------------------*/
static void closeftp(ftp_t* ftp) {
    tracet(NULL, 3, "closeftp: state=%d\n", ftp->state);
 
    if (ftp->state != 1) {
        free(ftp);
    }
}
/* read ftp ------------------------------------------------------------------*/
static int readftp(ftp_t* ftp, uint8_t* buff, int n, char* msg) {
    gtime_t time;
    uint8_t *p, *q;
 
    tracet(NULL, 4, "readftp: n=%d\n", n);
 
    time = utc2gpst(timeget());
 
    if (timediff(time, ftp->tnext) < 0.0) { /* until download time? */
        return 0;
    }
    if (ftp->state <= 0) { /* ftp/http not executed? */
        ftp->state = 1;
        sprintf(msg, "%s://%s", ftp->proto ? "http" : "ftp", ftp->addr);
 
        if (pthread_create(&ftp->thread, NULL, ftpthread, ftp)) {
            tracet(NULL, 2, "readftp: ftp thread create error\n");
            ftp->state = 3;
            strcpy(msg, "ftp thread error");
            return 0;
        }
    }
    if (ftp->state <= 1) {
        return 0; /* ftp/http on going? */
    }
 
    if (ftp->state == 3) { /* ftp error */
        sprintf(msg, "%s error (%d)", ftp->proto ? "http" : "ftp", ftp->error);
 
        /* set next retry time */
        ftp->tnext = nextdltime(ftp->topts, 0);
        ftp->state = 0;
        return 0;
    }
    /* return local file path if ftp completed */
    p = buff;
    q = (uint8_t*)ftp->local;
    while (*q && (int)(p - buff) < n) {
        *p++ = *q++;
    }
    p += sprintf((char*)p, "\r\n");
 
    /* set next download time */
    ftp->tnext = nextdltime(ftp->topts, 1);
    ftp->state = 0;
 
    strcpy(msg, "");
 
    return (int)(p - buff);
}
/* get state ftp -------------------------------------------------------------*/
static int stateftp(ftp_t* ftp) {
    if (!ftp) {
        return 0;
    } else if (ftp->state == 0) {
        return 2;
    } else if (ftp->state <= 2) {
        return 3;
    } else {
        return -1;
    }
}
/* get extended state ftp ----------------------------------------------------*/
static int statexftp(ftp_t* ftp, char* msg) {
    if (!ftp) {
        return 0;
    } else if (ftp->state == 0) {
        return 2;
    } else if (ftp->state <= 2) {
        return 3;
    } else {
        return -1;
    }
}
/* open memory buffer --------------------------------------------------------*/
static membuf_t* openmembuf(const char* path, char* msg) {
    membuf_t* membuf;
    int bufsize = DEFAULT_MEMBUF_SIZE;
 
    tracet(NULL, 3, "openmembuf: path=%s\n", path);
 
    msg[0] = '\0';
 
    sscanf(path, "%d", &bufsize);
 
    if (!(membuf = (membuf_t*)malloc(sizeof(membuf_t)))) {
        return NULL;
    }
    membuf->state = 1;
    membuf->rp = 0;
    membuf->wp = 0;
    if (!(membuf->buf = (uint8_t*)malloc(bufsize))) {
        free(membuf);
        return NULL;
    }
    membuf->bufsize = bufsize;
    rtk_initlock(&membuf->lock);
 
    sprintf(msg, "membuf sizebuf=%d", bufsize);
 
    return membuf;
}
/* close memory buffer -------------------------------------------------------*/
static void closemembuf(membuf_t* membuf) {
    tracet(NULL, 3, "closemembufp\n");
 
    free(membuf->buf);
    free(membuf);
}
/* read memory buffer --------------------------------------------------------*/
static int readmembuf(membuf_t* membuf, uint8_t* buff, int n, char* msg) {
    int i, nr = 0;
 
    tracet(NULL, 4, "readmembuf: n=%d\n", n);
 
    if (!membuf) {
        return 0;
    }
 
    rtk_lock(&membuf->lock);
 
    for (i = membuf->rp; i != membuf->wp && nr < n; i++) {
        if (i >= membuf->bufsize) {
            i = 0;
            if (i == membuf->wp) {
                break;
            }
        }
        buff[nr++] = membuf->buf[i];
    }
    membuf->rp = i;
    rtk_unlock(&membuf->lock);
    return nr;
}
/* write memory buffer -------------------------------------------------------*/
static int writemembuf(membuf_t* membuf, uint8_t* buff, int n, char* msg) {
    int i;
 
    tracet(NULL, 3, "writemembuf: n=%d\n", n);
 
    if (!membuf) {
        return 0;
    }
 
    rtk_lock(&membuf->lock);
 
    for (i = 0; i < n; i++) {
        membuf->buf[membuf->wp++] = buff[i];
        if (membuf->wp >= membuf->bufsize) {
            membuf->wp = 0;
        }
        if (membuf->wp == membuf->rp) {
            strcpy(msg, "mem-buffer overflow");
            membuf->state = -1;
            rtk_unlock(&membuf->lock);
            return i + 1;
        }
    }
    rtk_unlock(&membuf->lock);
    return i;
}
/* get state memory buffer ---------------------------------------------------*/
static int statemembuf(membuf_t* membuf) { return !membuf ? 0 : membuf->state; }
/* get extended state memory buffer ------------------------------------------*/
static int statexmembuf(membuf_t* membuf, char* msg) {
    char* p = msg;
    int state = !membuf ? 0 : membuf->state;
 
    p += sprintf(p, "membuf:\n");
    p += sprintf(p, "  state   = %d\n", state);
    if (!state) {
        return 0;
    }
    p += sprintf(p, "  buffsize= %d\n", membuf->bufsize);
    p += sprintf(p, "  wp      = %d\n", membuf->wp);
    p += sprintf(p, "  rp      = %d\n", membuf->rp);
    return state;
}
/* initialize stream environment -----------------------------------------------
 * initialize stream environment
 * args   : none
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strinitcom(void) { tracet(NULL, 3, "strinitcom:\n"); }
/* initialize stream -----------------------------------------------------------
 * initialize stream struct
 * args   : stream_t *stream IO  stream
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strinit(stream_t* stream) {
    tracet(NULL, 3, "strinit:\n");
 
    stream->type = 0;
    stream->mode = 0;
    stream->state = 0;
    stream->inb = stream->inr = stream->outb = stream->outr = 0;
    stream->tick_i = stream->tick_o = stream->tact = stream->inbt = stream->outbt = 0;
    rtk_initlock(&stream->lock);
    stream->port = NULL;
    stream->path[0] = '\0';
    stream->msg[0] = '\0';
}
/* open stream -----------------------------------------------------------------
 *
 * open stream to read or write data from or to virtual devices.
 *
 * args   : stream_t *stream IO  stream
 *          int type         I   stream type
 *                                 STR_SERIAL   = serial device
 *                                 STR_FILE     = file (record and playback)
 *                                 STR_TCPSVR   = TCP server
 *                                 STR_TCPCLI   = TCP client
 *                                 STR_NTRIPSVR = NTRIP server
 *                                 STR_NTRIPCLI = NTRIP client
 *                                 STR_NTRIPC_S = NTRIP caster server
 *                                 STR_NTRIPC_C = NTRIP caster client
 *                                 STR_UDPSVR   = UDP server (read only)
 *                                 STR_UDPCLI   = UDP client (write only)
 *                                 STR_MEMBUF   = memory buffer (FIFO)
 *                                 STR_FTP      = download by FTP (raed only)
 *                                 STR_HTTP     = download by HTTP (raed only)
 *          int mode         I   stream mode (STR_MODE_???)
 *                                 STR_MODE_R   = read only
 *                                 STR_MODE_W   = write only
 *                                 STR_MODE_RW  = read and write
 *          char *path       I   stream path (see below)
 *
 * return : status (0:error,1:ok)
 *
 * notes  : see reference [1] for NTRIP
 *          STR_FTP/HTTP needs "wget" in command search paths
 *
 * stream path ([] options):
 *
 *   STR_SERIAL   port[:brate[:bsize[:parity[:stopb[:fctr[#port]]]]]]
 *                    port  = COM??  (windows)
 *                            tty??? (linuex, omit /dev/)
 *                    brate = bit rate     (bps)
 *                    bsize = bit size     (7|8)
 *                    parity= parity       (n|o|e)
 *                    stopb = stop bits    (1|2)
 *                    fctr  = flow control (off|rts)
 *                    port  = tcp server port to output received stream
 *
 *   STR_FILE     path[::T][::+start][::xseppd][::S=swap][::P={4|8}]
 *                    path  = file path
 *                            (can include keywords defined by )
 *                    ::T   = enable time tag
 *                    start = replay start offset (s)
 *                    speed = replay speed factor
 *                    swap  = output swap interval (hr) (0: no swap)
 *                    ::P={4|8} = file pointer size (4:32bit,8:64bit)
 *
 *   STR_TCPSVR   :port
 *                    port  = TCP server port to accept
 *
 *   STR_TCPCLI   addr:port
 *                    addr  = TCP server address to connect
 *                    port  = TCP server port to connect
 *
 *   STR_NTRIPSVR [:passwd@]addr[:port]/mponit[:string]
 *                    addr  = NTRIP caster address to connect
 *                    port  = NTRIP caster server port to connect
 *                    passwd= NTRIP caster server password to connect
 *                    mpoint= NTRIP mountpoint
 *                    string= NTRIP server string
 *
 *   STR_NTRIPCLI [user[:passwd]@]addr[:port]/mpoint
 *                    addr  = NTRIP caster address to connect
 *                    port  = NTRIP caster client port to connect
 *                    user  = NTRIP caster client user to connect
 *                    passwd= NTRIP caster client password to connect
 *                    mpoint= NTRIP mountpoint
 *
 *   STR_NTRIPCAS [user[:passwd]@][:port]/mpoint[:srctbl]
 *                    port  = NTRIP caster client port to accept connection
 *                    user  = NTRIP caster client user to accept connection
 *                    passwd= NTRIP caster client password to accept connection
 *                    mpoint= NTRIP mountpoint
 *                    srctbl= NTRIP source table entry (STR) (ref [3] 6.3)
 *                      (ID;format;format-details;carrier;nav-system;network;
 *                       country;latitude;longitude;nmea;solution;generator;
 *                       compr-encrp;autentication;fee;bitrate;...;misc)
 *
 *   STR_UDPSVR   :port
 *                    port  = UDP server port to receive
 *
 *   STR_UDPCLI   addr:port
 *                    addr  = UDP server or broadcast address to send
 *                    port  = UDP server or broadcast port to send
 *
 *   STR_MEMBUF   [size]
 *                    size  = FIFO size (bytes) ("":4096)
 *
 *   STR_FTP      [user[:passwd]@]addr/path[::T=poff[,tint[,toff,tret]]]]
 *                    user  = FTP server user
 *                    passwd= FTP server password
 *                    addr  = FTP server address
 *                    path  = FTP server file path
 *                    poff  = time offset for path extension (s)
 *                    tint  = download interval (s)
 *                    toff  = download time offset (s)
 *                    tret  = download retry interval (s) (0:no retry)
 *
 *   STR_HTTP     addr/path[::T=poff[,tint[,toff,tret]]]]
 *                    addr  = HTTP server address
 *                    path  = HTTP server file path
 *                    poff  = time offset for path extension (s)
 *                    tint  = download interval (s)
 *                    toff  = download time offset (s)
 *                    tret  = download retry interval (s) (0:no retry)
 *
 *-----------------------------------------------------------------------------*/
extern int stropen(stream_t* stream, int type, int mode, const char* path) {
    tracet(NULL, 3, "stropen: type=%d mode=%d path=%s\n", type, mode, path);
 
    stream->type = type;
    stream->mode = mode;
    strcpy(stream->path, path);
    stream->inb = stream->inr = stream->outb = stream->outr = 0;
    stream->tick_i = stream->tick_o = tickget();
    stream->inbt = stream->outbt = 0;
    stream->msg[0] = '\0';
    stream->port = NULL;
    switch (type) {
        case STR_SERIAL:
            stream->port = openserial(path, mode, stream->msg);
            break;
        case STR_FILE:
            stream->port = openfile(path, mode, stream->msg);
            break;
        case STR_TCPSVR:
            stream->port = opentcpsvr(path, stream->msg);
            break;
        case STR_TCPCLI:
            stream->port = opentcpcli(path, stream->msg);
            break;
        case STR_NTRIPSVR:
            stream->port = openntrip(path, 0, stream->msg);
            break;
        case STR_NTRIPCLI:
            stream->port = openntrip(path, 1, stream->msg);
            break;
        case STR_NTRIPCAS:
            stream->port = openntripc(path, stream->msg);
            break;
        case STR_UDPSVR:
            stream->port = openudpsvr(path, stream->msg);
            break;
        case STR_UDPCLI:
            stream->port = openudpcli(path, stream->msg);
            break;
        case STR_MEMBUF:
            stream->port = openmembuf(path, stream->msg);
            break;
        case STR_FTP:
            stream->port = openftp(path, 0, stream->msg);
            break;
        case STR_HTTP:
            stream->port = openftp(path, 1, stream->msg);
            break;
        default:
            stream->state = 0;
            return 1;
    }
    stream->state = !stream->port ? -1 : 1;
    return stream->port != NULL;
}
/* close stream ----------------------------------------------------------------
 * close stream
 * args   : stream_t *stream IO  stream
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strclose(stream_t* stream) {
    tracet(NULL, 3, "strclose: type=%d mode=%d\n", stream->type, stream->mode);
 
    strlock(stream);
 
    if (stream->port) {
        switch (stream->type) {
            case STR_SERIAL:
                closeserial((serial_t*)stream->port);
                break;
            case STR_FILE:
                closefile((file_t*)stream->port);
                break;
            case STR_TCPSVR:
                closetcpsvr((tcpsvr_t*)stream->port);
                break;
            case STR_TCPCLI:
                closetcpcli((tcpcli_t*)stream->port);
                break;
            case STR_NTRIPSVR:
                closentrip((ntrip_t*)stream->port);
                break;
            case STR_NTRIPCLI:
                closentrip((ntrip_t*)stream->port);
                break;
            case STR_NTRIPCAS:
                closentripc((ntripc_t*)stream->port);
                break;
            case STR_UDPSVR:
                closeudpsvr((udp_t*)stream->port);
                break;
            case STR_UDPCLI:
                closeudpcli((udp_t*)stream->port);
                break;
            case STR_MEMBUF:
                closemembuf((membuf_t*)stream->port);
                break;
            case STR_FTP:
                closeftp((ftp_t*)stream->port);
                break;
            case STR_HTTP:
                closeftp((ftp_t*)stream->port);
                break;
        }
    } else {
        trace(NULL, 3, "no port to close stream: type=%d\n", stream->type);
    }
    stream->type = 0;
    stream->mode = 0;
    stream->state = 0;
    stream->inr = stream->outr = 0;
    stream->path[0] = '\0';
    stream->msg[0] = '\0';
    stream->port = NULL;
 
    strunlock(stream);
}
/* sync streams ----------------------------------------------------------------
 * sync time for streams
 * args   : stream_t *stream1 IO stream 1
 *          stream_t *stream2 IO stream 2
 * return : none
 * notes  : for replay files with time tags
 *-----------------------------------------------------------------------------*/
extern void strsync(stream_t* stream1, stream_t* stream2) {
    file_t *file1, *file2;
    if (stream1->type != STR_FILE || stream2->type != STR_FILE) {
        return;
    }
    file1 = (file_t*)stream1->port;
    file2 = (file_t*)stream2->port;
    if (file1 && file2) {
        syncfile(file1, file2);
    }
}
/* lock/unlock stream ----------------------------------------------------------
 * lock/unlock stream
 * args   : stream_t *stream I  stream
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strlock(stream_t* stream) { rtk_lock(&stream->lock); }
extern void strunlock(stream_t* stream) { rtk_unlock(&stream->lock); }
 
/* read stream -----------------------------------------------------------------
 * read data from stream (unblocked)
 * args   : stream_t *stream I  stream
 *          unsinged char *buff O data buffer
 *          int    n         I  maximum data length
 * return : read data length
 * notes  : if no data, return immediately with no data
 *-----------------------------------------------------------------------------*/
extern int strread(stream_t* stream, uint8_t* buff, int n) {
    uint32_t tick = tickget();
    char* msg = stream->msg;
    int nr = 0, tt;
 
    tracet(NULL, 4, "strread: n=%d\n", n);
 
    if (!(stream->mode & STR_MODE_R) || !stream->port) {
        return 0;
    }
 
    strlock(stream);
 
    switch (stream->type) {
        case STR_SERIAL:
            nr = readserial((serial_t*)stream->port, buff, n, msg);
            break;
        case STR_FILE:
            nr = readfile((file_t*)stream->port, buff, n, msg);
            break;
        case STR_TCPSVR:
            nr = readtcpsvr((tcpsvr_t*)stream->port, buff, n, msg);
            break;
        case STR_TCPCLI:
            nr = readtcpcli((tcpcli_t*)stream->port, buff, n, msg);
            break;
        case STR_NTRIPSVR:
        case STR_NTRIPCLI:
            nr = readntrip((ntrip_t*)stream->port, buff, n, msg);
            break;
        case STR_NTRIPCAS:
            nr = readntripc((ntripc_t*)stream->port, buff, n, msg);
            break;
        case STR_UDPSVR:
            nr = readudpsvr((udp_t*)stream->port, buff, n, msg);
            break;
        case STR_MEMBUF:
            nr = readmembuf((membuf_t*)stream->port, buff, n, msg);
            break;
        case STR_FTP:
            nr = readftp((ftp_t*)stream->port, buff, n, msg);
            break;
        case STR_HTTP:
            nr = readftp((ftp_t*)stream->port, buff, n, msg);
            break;
        default:
            strunlock(stream);
            return 0;
    }
    if (nr > 0) {
        stream->inb += nr;
        stream->tact = tick;
    }
    tt = (int)(tick - stream->tick_i);
    if (tt >= tirate) {
        stream->inr = (uint32_t)((double)((stream->inb - stream->inbt) * 8) / (tt * 0.001));
        stream->tick_i = tick;
        stream->inbt = stream->inb;
    }
    strunlock(stream);
    return nr;
}
/* write stream ----------------------------------------------------------------
 * write data to stream (unblocked)
 * args   : stream_t *stream I   stream
 *          unsinged char *buff I data buffer
 *          int    n         I   data length
 * return : bytes written (0: nothing written or error; -1: error for some types)
 * notes  : write data to buffer and return immediately
 *-----------------------------------------------------------------------------*/
extern int strwrite(stream_t* stream, uint8_t* buff, int n) {
    uint32_t tick = tickget();
    char* msg = stream->msg;
    int ns, tt;
 
    tracet(NULL, 4, "strwrite: n=%d\n", n);
 
    if (!(stream->mode & STR_MODE_W) || !stream->port) {
        return 0;
    }
 
    strlock(stream);
 
    switch (stream->type) {
        case STR_SERIAL:
            ns = writeserial((serial_t*)stream->port, buff, n, msg);
            break;
        case STR_FILE:
            ns = writefile((file_t*)stream->port, buff, n, msg);
            break;
        case STR_TCPSVR:
            ns = writetcpsvr((tcpsvr_t*)stream->port, buff, n, msg);
            break;
        case STR_TCPCLI:
            ns = writetcpcli((tcpcli_t*)stream->port, buff, n, msg);
            break;
        case STR_NTRIPSVR:
        case STR_NTRIPCLI:
            ns = writentrip((ntrip_t*)stream->port, buff, n, msg);
            break;
        case STR_NTRIPCAS:
            ns = writentripc((ntripc_t*)stream->port, buff, n, msg);
            break;
        case STR_UDPCLI:
            ns = writeudpcli((udp_t*)stream->port, buff, n, msg);
            break;
        case STR_MEMBUF:
            ns = writemembuf((membuf_t*)stream->port, buff, n, msg);
            break;
        case STR_FTP:
        case STR_HTTP:
        default:
            strunlock(stream);
            return 0;
    }
    if (ns > 0) {
        stream->outb += ns;
        stream->tact = tick;
    }
    tt = (int)(tick - stream->tick_o);
    if (tt > tirate) {
        stream->outr = (uint32_t)((double)((stream->outb - stream->outbt) * 8) / (tt * 0.001));
        stream->tick_o = tick;
        stream->outbt = stream->outb;
    }
    strunlock(stream);
    return ns;
}
/* get stream status -----------------------------------------------------------
 * get stream status
 * args   : stream_t *stream I   stream
 *          char   *msg      IO  status message (NULL: no output)
 * return : status (-1:error,0:close,1:wait,2:connect,3:active)
 *-----------------------------------------------------------------------------*/
extern int strstat(stream_t* stream, char* msg) {
    int state = 0;
 
    tracet(NULL, 4, "strstat:\n");
 
    strlock(stream);
    if (msg) {
        strncpy(msg, stream->msg, MAXSTRMSG - 1);
        msg[MAXSTRMSG - 1] = '\0';
    }
    if (!stream->port) {
        strunlock(stream);
        return stream->state;
    }
    switch (stream->type) {
        case STR_SERIAL:
            state = stateserial((serial_t*)stream->port);
            break;
        case STR_FILE:
            state = statefile((file_t*)stream->port);
            break;
        case STR_TCPSVR:
            state = statetcpsvr((tcpsvr_t*)stream->port);
            break;
        case STR_TCPCLI:
            state = statetcpcli((tcpcli_t*)stream->port);
            break;
        case STR_NTRIPSVR:
        case STR_NTRIPCLI:
            state = statentrip((ntrip_t*)stream->port);
            break;
        case STR_NTRIPCAS:
            state = statentripc((ntripc_t*)stream->port);
            break;
        case STR_UDPSVR:
            state = stateudpsvr((udp_t*)stream->port);
            break;
        case STR_UDPCLI:
            state = stateudpcli((udp_t*)stream->port);
            break;
        case STR_MEMBUF:
            state = statemembuf((membuf_t*)stream->port);
            break;
        case STR_FTP:
            state = stateftp((ftp_t*)stream->port);
            break;
        case STR_HTTP:
            state = stateftp((ftp_t*)stream->port);
            break;
        default:
            strunlock(stream);
            return 0;
    }
    if (state == 2 && (int)(tickget() - stream->tact) <= TINTACT) {
        state = 3;
    }
    strunlock(stream);
    return state;
}
/* get extended stream status ---------------------------------------------------
 * get extended stream status
 * args   : stream_t *stream I   stream
 *          char   *msg      IO  extended status message
 * return : status (-1:error,0:close,1:wait,2:connect,3:active)
 *-----------------------------------------------------------------------------*/
extern int strstatx(stream_t* stream, char* msg) {
    int state = 0;
 
    tracet(NULL, 4, "strstatx:\n");
 
    strlock(stream);
 
    if (!stream->port) {
        strunlock(stream);
        return stream->state;
    }
    switch (stream->type) {
        case STR_SERIAL:
            state = statexserial((serial_t*)stream->port, msg);
            break;
        case STR_FILE:
            state = statexfile((file_t*)stream->port, msg);
            break;
        case STR_TCPSVR:
            state = statextcpsvr((tcpsvr_t*)stream->port, msg);
            break;
        case STR_TCPCLI:
            state = statextcpcli((tcpcli_t*)stream->port, msg);
            break;
        case STR_NTRIPSVR:
        case STR_NTRIPCLI:
            state = statexntrip((ntrip_t*)stream->port, msg);
            break;
        case STR_NTRIPCAS:
            state = statexntripc((ntripc_t*)stream->port, msg);
            break;
        case STR_UDPSVR:
            state = statexudpsvr((udp_t*)stream->port, msg);
            break;
        case STR_UDPCLI:
            state = statexudpcli((udp_t*)stream->port, msg);
            break;
        case STR_MEMBUF:
            state = statexmembuf((membuf_t*)stream->port, msg);
            break;
        case STR_FTP:
            state = statexftp((ftp_t*)stream->port, msg);
            break;
        case STR_HTTP:
            state = statexftp((ftp_t*)stream->port, msg);
            break;
        default:
            msg[0] = '\0';
            strunlock(stream);
            return 0;
    }
    if (state == 2 && (int)(tickget() - stream->tact) <= TINTACT) {
        state = 3;
    }
    strunlock(stream);
    return state;
}
/* get stream statistics summary -----------------------------------------------
 * get stream statistics summary
 * args   : stream_t *stream I   stream
 *          uint64_t *inb    IO   bytes of input  (NULL: no output)
 *          int    *inr      IO   bps of input    (NULL: no output)
 *          uint64_t *outb   IO   bytes of output (NULL: no output)
 *          int    *outr     IO   bps of output   (NULL: no output)
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsum(stream_t* stream, uint64_t* inb, int* inr, uint64_t* outb, int* outr) {
    tracet(NULL, 4, "strsum:\n");
 
    strlock(stream);
    if (inb) {
        *inb = stream->inb;
    }
    if (inr) {
        *inr = stream->inr;
    }
    if (outb) {
        *outb = stream->outb;
    }
    if (outr) {
        *outr = stream->outr;
    }
    strunlock(stream);
}
/* set global stream options ---------------------------------------------------
 * set global stream options
 * args   : int    *opt      I   options
 *              opt[0]= inactive timeout (ms) (0: no timeout)
 *              opt[1]= interval to reconnect (ms)
 *              opt[2]= averaging time of data rate (ms)
 *              opt[3]= receive/send buffer size (bytes);
 *              opt[4]= file swap margin (s)
 *              opt[5]= reserved
 *              opt[6]= reserved
 *              opt[7]= reserved
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsetopt(const int* opt) {
    tracet(NULL, 3, "strsetopt: opt=%d %d %d %d %d %d %d %d\n", opt[0], opt[1], opt[2], opt[3], opt[4], opt[5], opt[6],
           opt[7]);
 
    toinact = 0 < opt[0] && opt[0] < 1000 ? 1000 : opt[0]; /* >=1s */
    ticonnect = opt[1] < 1000 ? 1000 : opt[1];             /* >=1s */
    tirate = opt[2] < 100 ? 100 : opt[2];                  /* >=0.1s */
    buffsize = opt[3] < 4096 ? 4096 : opt[3];              /* >=4096byte */
    fswapmargin = opt[4] < 0 ? 0 : opt[4];
}
/* set timeout time ------------------------------------------------------------
 * set timeout time
 * args   : stream_t *stream I   stream (STR_TCPCLI,STR_NTRIPCLI,STR_NTRIPSVR)
 *          int     toinact  I   inactive timeout (ms) (0: no timeout)
 *          int     tirecon  I   reconnect interval (ms) (0: no reconnect)
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsettimeout(stream_t* stream, int toinact, int tirecon) {
    tcpcli_t* tcpcli;
 
    tracet(NULL, 3, "strsettimeout: toinact=%d tirecon=%d\n", toinact, tirecon);
 
    if (stream->type == STR_TCPCLI) {
        tcpcli = (tcpcli_t*)stream->port;
    } else if (stream->type == STR_NTRIPCLI || stream->type == STR_NTRIPSVR) {
        tcpcli = ((ntrip_t*)stream->port)->tcp;
    } else {
        return;
    }
 
    tcpcli->toinact = toinact;
    tcpcli->tirecon = tirecon;
}
/* set local directory ---------------------------------------------------------
 * set local directory path for ftp/http download
 * args   : char   *dir      I   directory for download files
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsetdir(const char* dir) {
    tracet(NULL, 3, "strsetdir: dir=%s\n", dir);
 
    strcpy(localdir, dir);
}
/* set http/ntrip proxy address ------------------------------------------------
 * set http/ntrip proxy address
 * args   : char   *addr     I   http/ntrip proxy address <address>:<port>
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsetproxy(const char* addr) {
    tracet(NULL, 3, "strsetproxy: addr=%s\n", addr);
 
    strcpy(proxyaddr, addr);
}
/* set default ntrip version ---------------------------------------------------
 * set default NTRIP version for new connections
 * args   : int ver     I   NTRIP version (0:auto, 1:v1, 2:v2)
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsetntripver(int ver) {
    tracet(NULL, 3, "strsetntripver: ver=%d\n", ver);
 
    if (ver >= 0 && ver <= 2) {
        ntrip_ver_default = ver;
    }
}
/* get stream time -------------------------------------------------------------
 * get stream time
 * args   : stream_t *stream I   stream
 * return : current time or replay time for playback file
 *-----------------------------------------------------------------------------*/
extern gtime_t strgettime(stream_t* stream) {
    file_t* file;
    if (stream->type == STR_FILE && (stream->mode & STR_MODE_R) && (file = (file_t*)stream->port)) {
        return timeadd(file->time, file->start); /* replay start time */
    }
    return utc2gpst(timeget());
}
/* send nmea request -----------------------------------------------------------
 * send nmea gpgga message to stream
 * args   : stream_t *stream I   stream
 *          sol_t *sol       I   solution
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsendnmea(stream_t* stream, const sol_t* sol) {
    uint8_t buff[1024];
    int n;
 
    tracet(NULL, 3, "strsendnmea: rr=%.3f %.3f %.3f\n", sol->rr[0], sol->rr[1], sol->rr[2]);
 
    n = outnmea_gga(buff, sol);
    strwrite(stream, buff, n);
}
/* generate general hex message ----------------------------------------------*/
static int gen_hex(const char* msg, uint8_t* buff) {
    uint8_t* q = buff;
    char mbuff[1024] = "", *args[256], *p;
    uint32_t byte;
    int i, narg = 0;
 
    trace(NULL, 4, "gen_hex: msg=%s\n", msg);
 
    strncpy(mbuff, msg, 1023);
    for (p = strtok(mbuff, " "); p && narg < 256; p = strtok(NULL, " ")) {
        args[narg++] = p;
    }
    for (i = 0; i < narg; i++) {
        if (sscanf(args[i], "%x", &byte)) {
            *q++ = (uint8_t)byte;
        }
    }
    return (int)(q - buff);
}
/* set bitrate ---------------------------------------------------------------*/
static int set_brate(stream_t* str, int brate) {
    char path[1024], buff[1024] = "", *p, *q;
    int type = str->type, mode = str->mode;
 
    if (type != STR_SERIAL) {
        return 0;
    }
 
    strcpy(path, str->path);
 
    if (!(p = strchr(path, ':'))) {
        sprintf(path + strlen(path), ":%d", brate);
    } else {
        if ((q = strchr(p + 1, ':'))) {
            strcpy(buff, q);
        }
        sprintf(p, ":%d%s", brate, buff);
    }
    strclose(str);
    return stropen(str, type, mode, path);
}
/* send receiver command -------------------------------------------------------
 * send receiver commands to stream
 * args   : stream_t *stream I   stream
 *          char   *cmd      I   receiver command strings
 * return : none
 *-----------------------------------------------------------------------------*/
extern void strsendcmd(stream_t* str, const char* cmd) {
    uint8_t buff[1024];
    const char *p = cmd, *q;
    char msg[1024], cmdend[] = "\r\n";
    int n, m, ms, brate;
 
    tracet(NULL, 3, "strsendcmd: cmd=%s\n", cmd);
 
    while (1) {
        for (q = p;; q++) {
            if (*q == '\r' || *q == '\n' || *q == '\0') {
                break;
            }
        }
        n = (int)(q - p);
        strncpy(msg, p, n);
        msg[n] = '\0';
 
        if (!*msg || *msg == '#') { /* null or comment */
            /* skip */
        } else if (*msg == '!') { /* binary escape */
 
            if (!strncmp(msg + 1, "WAIT", 4)) { /* wait */
                if (sscanf(msg + 5, "%d", &ms) < 1) {
                    ms = 100;
                }
                if (ms > 3000) {
                    ms = 3000; /* max 3 s */
                }
                sleepms(ms);
            } else if (!strncmp(msg + 1, "BRATE", 5)) { /* set bitrate */
                if (sscanf(msg + 6, "%d", &brate) < 1) {
                    brate = 9600;
                }
                set_brate(str, brate);
                sleepms(500);
            } else if (!strncmp(msg + 1, "UBX", 3)) { /* ublox */
                if ((m = gen_ubx(msg + 4, buff)) > 0) {
                    strwrite(str, buff, m);
                }
            } else if (!strncmp(msg + 1, "STQ", 3)) { /* skytraq */
                if ((m = gen_stq(msg + 4, buff)) > 0) {
                    strwrite(str, buff, m);
                }
            } else if (!strncmp(msg + 1, "NVS", 3)) { /* nvs */
                if ((m = gen_nvs(msg + 4, buff)) > 0) {
                    strwrite(str, buff, m);
                }
            } else if (!strncmp(msg + 1, "HEX", 3)) { /* general hex message */
                if ((m = gen_hex(msg + 4, buff)) > 0) {
                    strwrite(str, buff, m);
                }
            }
        } else {
            strcat(msg, cmdend);
            strwrite(str, (uint8_t*)msg, n + 2);
        }
        if (*q == '\0') {
            break;
        } else {
            p = q + 1;
        }
    }
}