src/client.c

   1 /*
   2  * C Client for the DX Spider cluster program
   3  *
   4  * Eventually this program will be a complete replacement
   5  * for the perl version.
   6  *
   7  * This program provides the glue necessary to talk between
   8  * an input (eg from telnet or ax25) and the perl DXSpider
   9  * node.
  10  *
  11  * Currently, this program connects STDIN/STDOUT to the
  12  * message system used by cluster.pl
  13  *
  14  * Copyright (c) 2000 Dirk Koopman G1TLH
  15  *
  16  * $Id$
  17  */
  18
  19 #include <stdio.h>
  20 #include <sys/time.h>
  21 #include <sys/types.h>
  22 #include <unistd.h>
  23 #include <ctype.h>
  24 #include <stdlib.h>
  25 #include <stdarg.h>
  26 #include <netdb.h>
  27 #include <sys/socket.h>
  28 #include <netinet/in.h>
  29 #include <errno.h>
  30 #include <signal.h>
  31 #include <string.h>
  32
  33 #include "sel.h"
  34 #include "cmsg.h"
  35
  36 #define TEXT 1
  37 #define MSG 2
  38 #define MAXBUFL 1024
  39
  40 typedef struct
  41 {
  42         int cnum;                                       /* the connection number */
  43         int sort;                                       /* the type of connection either text or msg */
  44         cmsg_t *in;                                     /* current input message being built up */
  45         cmsg_t *out;                            /* current output message being sent */
  46         reft *inq;                                      /* input queue */
  47         reft *outq;                                     /* output queue */
  48         sel_t *sp;                                      /* my select fcb address */
  49 } fcb_t;
  50
  51 char *node_addr = "localhost";  /* the node tcp address */
  52 int node_port = 27754;                  /* the tcp port of the node at the above address */
  53 char *call;                                             /* the caller's callsign */
  54 char *connsort;                                 /* the type of connection */
  55 fcb_t *in;                                              /* the fcb of 'stdin' that I shall use */
  56 fcb_t *out;                                             /* the fcb of 'stdout' that I shall use */
  57 fcb_t *node;                                    /* the fcb of the msg system */
  58 char nl = '\n';                                 /* line end character */
  59 char ending = 0;                                /* set this to end the program */
  60 char send_Z = 1;                                /* set a Z record to the node on termination */
  61
  62 /*
  63  * utility routines - various
  64  */
  65
  66 void die(char *s, ...)
  67 {
  68         char buf[2000];
  69
  70         va_list ap;
  71         va_start(ap, s);
  72         vsprintf(buf, s, ap);
  73         va_end(ap);
  74         fprintf(stderr, buf);
  75         exit(-1);
  76 }
  77
  78 char *strupper(char *s)
  79 {
  80         char *d = malloc(strlen(s)+1);
  81         char *p = d;
  82
  83         if (!d)
  84                 die("out of room in strupper");
  85         while (*p++ = toupper(*s++)) ;
  86         return d;
  87 }
  88
  89 char *strlower(char *s)
  90 {
  91         char *d = malloc(strlen(s)+1);
  92         char *p = d;
  93
  94         if (!d)
  95                 die("out of room in strlower");
  96         while (*p++ = tolower(*s++)) ;
  97         return d;
  98 }
  99
 100 int eq(char *a, char *b)
 101 {
 102         return (strcmp(a, b) == 0);
 103 }
 104
 105 /*
 106  * higher level send and receive routines
 107  */
 108
 109 fcb_t *fcb_new(int cnum, int sort)
 110 {
 111         fcb_t *f = malloc(sizeof(fcb_t));
 112         if (!f)
 113                 die("no room in fcb_new");
 114         memset (f, 0, sizeof(fcb_t));
 115         f->cnum = cnum;
 116         f->sort = sort;
 117         f->inq = chain_new();
 118         f->outq = chain_new();
 119         return f;
 120 }
 121
 122 void send_text(fcb_t *f, char *s, int l)
 123 {
 124         cmsg_t *mp;
 125         mp = cmsg_new(l+1, f->sort, f);
 126         memcpy(mp->inp, s, l);
 127         mp->inp += l;
 128         *mp->inp++ = nl;
 129         cmsg_send(f->outq, mp, 0);
 130         f->sp->flags |= SEL_OUTPUT;
 131 }
 132
 133 void send_msg(fcb_t *f, char let, char *s, int l)
 134 {
 135         cmsg_t *mp;
 136         int ln;
 137         int myl = strlen(call)+2+l;
 138
 139         mp = cmsg_new(myl+4, f->sort, f);
 140         ln = htonl(myl);
 141         memcpy(mp->inp, &ln, 4);
 142         mp->inp += 4;
 143         *mp->inp++ = let;
 144         strcpy(mp->inp, call);
 145         mp->inp += strlen(call);
 146         *mp->inp++ = '|';
 147         if (l) {
 148                 memcpy(mp->inp, s, l);
 149                 mp->inp += l;
 150         }
 151         *mp->inp = 0;
 152         cmsg_send(f->outq, mp, 0);
 153         f->sp->flags |= SEL_OUTPUT;
 154 }
 155
 156 int fcb_handler(sel_t *sp, int in, int out, int err)
 157 {
 158         fcb_t *f = sp->fcb;
 159         cmsg_t *mp;
 160
 161         /* input modes */
 162         if (in) {
 163                 char *p, buf[MAXBUFL];
 164                 int r;
 165
 166                 /* read what we have into a buffer */
 167                 r = read(f->cnum, buf, MAXBUFL);
 168                 if (r < 0) {
 169                         switch (errno) {
 170                         case EINTR:
 171                         case EINPROGRESS:
 172                         case EAGAIN:
 173                                 goto lout;
 174                         default:
 175                                 if (f->sort == MSG)
 176                                         send_Z = 0;
 177                                 ending++;
 178                                 return 0;
 179                         }
 180                 } else if (r == 0) {
 181                         if (f->sort == MSG)
 182                                 send_Z = 0;
 183                         ending++;
 184                         return 0;
 185                 }
 186
 187                 /* create a new message buffer if required */
 188                 if (!f->in)
 189                         f->in = cmsg_new(MAXBUFL, sp->sort, f);
 190                 mp = f->in;
 191
 192                 switch (f->sort) {
 193                 case TEXT:
 194                         p = buf;
 195                         while (r > 0 && p < &buf[r]) {
 196
 197                                 /*
 198                                  * if we have a nl then send the message upstairs
 199                                  * start a new message
 200                                  */
 201
 202                                 if (*p == nl) {
 203                                         if (mp->inp == mp->data)
 204                                                 *mp->inp++ = ' ';
 205                                         *mp->inp = 0;              /* zero terminate it, but don't include it in the length */
 206                                         cmsg_send(f->inq, mp, 0);
 207                                         f->in = mp = cmsg_new(MAXBUFL, sp->sort, f);
 208                                         ++p;
 209                                 } else {
 210                                         if (mp->inp < &mp->data[MAXBUFL])
 211                                                 *mp->inp++ = *p++;
 212                                         else {
 213                                                 mp->inp = mp->data;
 214                                         }
 215                                 }
 216                         }
 217                         break;
 218
 219                 case MSG:
 220                         p = buf;
 221                         while (r > 0 && p < &buf[r]) {
 222
 223                                 /* build up the size into the likely message length (yes I know it's a short) */
 224                                 switch (mp->state) {
 225                                 case 0:
 226                                 case 1:
 227                                 case 2:
 228                                 case 3:
 229                                         mp->size = (mp->size << 8) | *p++;
 230                                         mp->state++;
 231                                         break;
 232                                 default:
 233                                         if (mp->inp - mp->data < mp->size) {
 234                                                 *mp->inp++ = *p++;
 235                                         } else {
 236                                                 /* kick it upstairs */
 237                                                 cmsg_send(f->inq, mp, 0);
 238                                                 mp = f->in = cmsg_new(MAXBUFL, f->sort, f);
 239                                         }
 240                                 }
 241                         }
 242                         break;
 243
 244                 default:
 245                         die("invalid sort (%d) in input handler", f->sort);
 246                 }
 247         }
 248
 249         /* output modes */
 250 lout:;
 251         if (out) {
 252                 int l, r;
 253
 254                 if (!f->out) {
 255                         mp = f->out = cmsg_next(f->outq);
 256                         if (!mp) {
 257                                 sp->flags &= ~SEL_OUTPUT;
 258                                 return 0;
 259                         }
 260                         mp->inp = mp->data;
 261                 }
 262                 l = mp->size - (mp->inp - mp->data);
 263                 if (l > 0) {
 264                         r = write(f->cnum, mp->inp, l);
 265                         if (r < 0) {
 266                                 switch (errno) {
 267                                 case EINTR:
 268                                 case EINPROGRESS:
 269                                 case EAGAIN:
 270                                         goto lend;
 271                                 default:
 272                                         if (f->sort == MSG)
 273                                                 send_Z = 0;
 274                                         ending++;
 275                                         return;
 276                                 }
 277                         } else if (r > 0) {
 278                                 mp->inp += r;
 279                         }
 280                 } else if (l < 0)
 281                         die("got negative length in handler on node");
 282                 if (mp->inp - mp->data >= mp->size) {
 283                         cmsg_callback(mp, 0);
 284                         f->out = 0;
 285                         if (!is_chain_empty(f->outq))
 286                                 sp->flags &= ~SEL_OUTPUT;
 287                 }
 288         }
 289 lend:;
 290         return 0;
 291 }
 292
 293 /*
 294  * things to do with initialisation
 295  */
 296
 297 void initargs(int argc, char *argv[])
 298 {
 299         int i;
 300         if (argc >= 2) {
 301                 call = strupper(argv[1]);
 302                 if (eq(call, "LOGIN"))
 303                         die("login not implemented (yet)");
 304         }
 305         if (!call)
 306                 die("Must have at least a callsign (for now)");
 307
 308         if (argc >= 3) {
 309                 connsort = strlower(argv[2]);
 310                 if (eq(connsort, "telnet") || eq(connsort, "local")) {
 311                         nl = '\n';
 312                 } else if (eq(connsort, "ax25")) {
 313                         nl = '\r';
 314                 } else {
 315                         die("2nd argument must be \"telnet\" or \"ax25\" or \"local\"");
 316                 }
 317         } else {
 318                 connsort = "local";
 319         }
 320 }
 321
 322 void connect_to_node()
 323 {
 324         struct hostent *hp, *gethostbyname();
 325         struct sockaddr_in server;
 326         int nodef;
 327         sel_t *sp;
 328
 329         if ((hp = gethostbyname(node_addr)) == 0)
 330                 die("Unknown host tcp host %s for printer", node_addr);
 331
 332         memset(&server, 0, sizeof server);
 333         server.sin_family = AF_INET;
 334         memcpy(&server.sin_addr, hp->h_addr, hp->h_length);
 335         server.sin_port = htons(node_port);
 336
 337         nodef = socket(AF_INET, SOCK_STREAM, 0);
 338         if (nodef < 0)
 339                 die("Can't open socket to %s port %d (%d)", node_addr, node_port, errno);
 340
 341         if (connect(nodef, (struct sockaddr *) &server, sizeof server) < 0) {
 342                 die("Error on connect to %s port %d (%d)", node_addr, node_port, errno);
 343         }
 344         node = fcb_new(nodef, MSG);
 345         node->sp = sel_open(nodef, node, "Msg System", fcb_handler, MSG, SEL_INPUT);
 346
 347 }
 348
 349 /*
 350  * things to do with going away
 351  */
 352
 353 void term_timeout(int i)
 354 {
 355         /* none of this is going to be reused so don't bother cleaning up properly */
 356         if (out)
 357                 out = 0;
 358         if (node) {
 359                 close(node->cnum);
 360                 node = 0;
 361         }
 362         exit(i);
 363 }
 364
 365 void terminate(int i)
 366 {
 367         if (send_Z && call) {
 368                 send_msg(node, 'Z', "", 0);
 369         }
 370
 371         signal(SIGALRM, term_timeout);
 372         alarm(10);
 373
 374         while ((out && !is_chain_empty(out->outq)) ||
 375                    (node && !is_chain_empty(node->outq))) {
 376                 sel_run();
 377         }
 378         if (node)
 379                 close(node->cnum);
 380         exit(i);
 381 }
 382
 383 /*
 384  * things to do with ongoing processing of inputs
 385  */
 386
 387 void process_stdin()
 388 {
 389         cmsg_t *mp = cmsg_next(in->inq);
 390         if (mp) {
 391                 send_msg(node, 'I', mp->data, mp->size);
 392                 cmsg_callback(mp, 0);
 393         }
 394 }
 395
 396 void process_node()
 397 {
 398         cmsg_t *mp = cmsg_next(node->inq);
 399         if (mp) {
 400                 char *p = strchr(mp->data, '|');
 401                 if (p)
 402                         p++;
 403                 switch (mp->data[0]) {
 404                 case 'Z':
 405                         send_Z = 0;
 406                         ending++;
 407                         return;
 408                 case 'D':
 409                         if (p) {
 410                                 int l = mp->inp - (unsigned char *) p;
 411                                 send_text(out, p, l);
 412                         }
 413                         break;
 414                 default:
 415                         break;
 416                 }
 417                 cmsg_callback(mp, 0);
 418         }
 419         if (is_chain_empty(out->outq))
 420                 fsync(out->cnum);
 421 }
 422
 423 /*
 424  * the program itself....
 425  */
 426
 427 main(int argc, char *argv[])
 428 {
 429         initargs(argc, argv);
 430         sel_init(10, 0, 10000);
 431
 432         signal(SIGHUP, SIG_IGN);
 433
 434         signal(SIGINT, terminate);
 435         signal(SIGQUIT, terminate);
 436         signal(SIGTERM, terminate);
 437         signal(SIGPWR, terminate);
 438
 439         /* connect up stdin, stdout and message system */
 440         in = fcb_new(0, TEXT);
 441         in->sp = sel_open(0, in, "STDIN", fcb_handler, TEXT, SEL_INPUT);
 442         out = fcb_new(1, TEXT);
 443         out->sp = sel_open(1, out, "STDOUT", fcb_handler, TEXT, 0);
 444         connect_to_node();
 445
 446         /* tell the cluster who I am */
 447         send_msg(node, 'A', connsort, strlen(connsort));
 448
 449         /* main processing loop */
 450         while (!ending) {
 451                 sel_run();
 452                 if (!ending) {
 453                         process_stdin();
 454                         process_node();
 455                 }
 456         }
 457         terminate(0);
 458 }
 459
 460
 461
 462
 463