recv.c

   1 /*
   2  * Copyright (C) 2005-2006 Andre Noll <maan@systemlinux.org>
   3  *
   4  *     This program is free software; you can redistribute it and/or modify
   5  *     it under the terms of the GNU General Public License as published by
   6  *     the Free Software Foundation; either version 2 of the License, or
   7  *     (at your option) any later version.
   8  *
   9  *     This program is distributed in the hope that it will be useful,
  10  *     but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  *     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  *     GNU General Public License for more details.
  13  *
  14  *     You should have received a copy of the GNU General Public License
  15  *     along with this program; if not, write to the Free Software
  16  *     Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111, USA.
  17  */
  18 #include "para.h"
  19
  20 #include "list.h"
  21 #include "sched.h"
  22 #include "recv.h"
  23 #include "recv.cmdline.h"
  24 #include "fd.h"
  25 #include "error.h"
  26 #include "stdout.h"
  27
  28 struct gengetopt_args_info conf;
  29
  30 INIT_RECV_ERRLISTS;
  31
  32 __printf_2_3 void para_log(int ll, const char* fmt,...)
  33 {
  34         va_list argp;
  35
  36         /* ignore log message if loglevel is not high enough */
  37         if (ll < conf.loglevel_arg)
  38                 return;
  39         va_start(argp, fmt);
  40         vfprintf(stderr, fmt, argp);
  41         va_end(argp);
  42 }
  43
  44 static void *parse_config(int argc, char *argv[], int *receiver_num)
  45 {
  46         int i;
  47
  48         if (cmdline_parser(argc, argv, &conf))
  49                 return NULL;
  50         if (conf.list_receivers_given) {
  51                 printf("available receivers: ");
  52                 for (i = 0; receivers[i].name; i++)
  53                         printf("%s%s", i? " " : "", receivers[i].name);
  54                 printf("\nTry\n\tpara_recv -r '<receivername> -h'\n"
  55                         "for help on <receivername>.\n");
  56                 exit(EXIT_SUCCESS);
  57         }
  58         return check_receiver_arg(conf.receiver_arg, receiver_num);
  59 }
  60
  61 #if 0
  62 int main(int argc, char *argv[])
  63 {
  64         int ret, eof = 0, max, r_opened = 0, receiver_num;
  65         struct timeval timeout;
  66         struct  receiver *r = NULL;
  67         fd_set rfds, wfds;
  68         struct receiver_node rn;
  69
  70         memset(&rn, 0, sizeof(struct receiver_node));
  71         for (ret = 0; receivers[ret].name; ret++)
  72                 receivers[ret].init(&receivers[ret]);
  73         ret = -E_RECV_SYNTAX;
  74         rn.conf = parse_config(argc, argv, &receiver_num);
  75         if (!rn.conf) {
  76                 PARA_EMERG_LOG("%s", "parse failed\n");
  77                 goto out;
  78         }
  79         r = &receivers[receiver_num];
  80         rn.receiver = r;
  81         ret = r->open(&rn);
  82         if (ret < 0)
  83                 goto out;
  84         r_opened = 1;
  85 recv:
  86         FD_ZERO(&rfds);
  87         FD_ZERO(&wfds);
  88         timeout.tv_sec = 0;
  89         timeout.tv_usec = 999 * 1000;
  90         max = -1;
  91         ret = r->pre_select(&rn, &rfds, &wfds, &timeout);
  92         max = PARA_MAX(max, ret);
  93
  94         PARA_DEBUG_LOG("timeout: %lums, max: %d\n", tv2ms(&timeout), max);
  95         ret = para_select(max + 1, &rfds, &wfds, &timeout);
  96         if (ret < 0) {
  97                 ret = -E_RECV_SELECT;
  98                 goto out;
  99         }
 100         ret = r->post_select(&rn, ret, &rfds, &wfds);
 101         if (ret < 0)
 102                 goto out;
 103         if (!ret)
 104                 eof = 1;
 105         if (!rn.loaded) {
 106                 if (eof)
 107                         goto out;
 108                 goto recv;
 109         }
 110         ret = write(STDOUT_FILENO, rn.buf, rn.loaded);
 111         PARA_DEBUG_LOG("wrote %d/%zd\n", ret, rn.loaded);
 112         if (ret < 0) {
 113                 ret = -E_WRITE_STDOUT;
 114                 goto out;
 115         }
 116         if (ret != rn.loaded) {
 117                 PARA_INFO_LOG("short write %d/%zd\n", ret, rn.loaded);
 118                 memmove(rn.buf, rn.buf + ret, rn.loaded - ret);
 119         }
 120         rn.loaded -= ret;
 121         if (rn.loaded || !eof)
 122                 goto recv;
 123 out:
 124         if (r_opened)
 125                 r->close(&rn);
 126         if (r)
 127                 r->shutdown();
 128         if (ret < 0)
 129                 PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
 130         return ret;
 131 }
 132 #endif
 133
 134 void rn_event_handler(struct task *t)
 135 {
 136         PARA_ERROR_LOG("%s\n", PARA_STRERROR(-t->ret));
 137         unregister_task(t);
 138 }
 139
 140 void stdout_event_handler(struct task *t)
 141 {
 142         PARA_ERROR_LOG("%s\n", PARA_STRERROR(-t->ret));
 143         unregister_task(t);
 144 }
 145
 146 int main(int argc, char *argv[])
 147 {
 148         int ret, eof = 0, max, r_opened = 0, receiver_num;
 149         struct timeval timeout;
 150         struct  receiver *r = NULL;
 151         fd_set rfds, wfds;
 152         struct receiver_node rn;
 153         struct stdout_task sot;
 154         struct sched s;
 155
 156         init_sched();
 157         s.default_timeout.tv_sec = 1;
 158         s.default_timeout.tv_usec = 0;
 159
 160         memset(&rn, 0, sizeof(struct receiver_node));
 161         for (ret = 0; receivers[ret].name; ret++)
 162                 receivers[ret].init(&receivers[ret]);
 163         ret = -E_RECV_SYNTAX;
 164         rn.conf = parse_config(argc, argv, &receiver_num);
 165         if (!rn.conf) {
 166                 PARA_EMERG_LOG("%s", "parse failed\n");
 167                 goto out;
 168         }
 169         r = &receivers[receiver_num];
 170         rn.receiver = r;
 171         ret = r->open(&rn);
 172         if (ret < 0)
 173                 goto out;
 174         r_opened = 1;
 175
 176         sot.task.private_data = &sot;
 177         sot.task.pre_select = stdout_pre_select;
 178         sot.task.post_select = stdout_post_select;
 179         sot.task.event_handler = stdout_event_handler;
 180         sot.task.flags = 0;
 181         sprintf(sot.task.status, "stdout writer");
 182         sot.buf = rn.buf;
 183         sot.loaded = &rn.loaded;
 184         sot.eof = &rn.eof;
 185         register_task(&sot.task);
 186
 187         rn.task.private_data = &rn;
 188         rn.task.pre_select = r->pre_select;
 189         rn.task.post_select = r->post_select;
 190         rn.task.event_handler = rn_event_handler;
 191         rn.task.flags = 0;
 192         sprintf(rn.task.status, "receiver node");
 193         register_task(&rn.task);
 194
 195
 196         ret = sched(&s);
 197 out:
 198         if (r_opened)
 199                 r->close(&rn);
 200         if (r)
 201                 r->shutdown();
 202         if (ret < 0)
 203                 PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));
 204         return ret;
 205 }