src/engine.c

   1 /*
   2  * transsip - the telephony toolkit
   3  * By Daniel Borkmann <daniel@transsip.org>
   4  * Copyright 2011, 2012 Daniel Borkmann <dborkma@tik.ee.ethz.ch>
   5  * Subject to the GPL, version 2.
   6  */
   7
   8 #include <sys/types.h>
   9 #include <sys/stat.h>
  10 #include <fcntl.h>
  11 #include <pthread.h>
  12 #include <string.h>
  13 #include <signal.h>
  14 #include <unistd.h>
  15 #include <celt/celt.h>
  16 #include <speex/speex_jitter.h>
  17 #include <speex/speex_echo.h>
  18 #include <speex/speex_preprocess.h>
  19 #include <netinet/in.h>
  20 #include <arpa/inet.h>
  21 #include <netdb.h>
  22 #include <assert.h>
  23
  24 #include "built_in.h"
  25 #include "alsa.h"
  26 #include "die.h"
  27 #include "xmalloc.h"
  28 #include "xutils.h"
  29 #include "call_notifier.h"
  30
  31 #define SAMPLING_RATE   48000
  32 #define FRAME_SIZE      256
  33 #define PACKETSIZE      43
  34 #define MAX_MSG         1500
  35 #define PATH_MAX        512
  36
  37 struct transsip_hdr {
  38         uint32_t seq;
  39         __extension__ uint8_t est:1,
  40                               psh:1,
  41                               bsy:1,
  42                               fin:1,
  43                               res1:4;
  44 } __attribute__((packed));
  45
  46 enum engine_state_num {
  47         ENGINE_STATE_IDLE = CALL_STATE_MACHINE_IDLE,
  48         ENGINE_STATE_CALLOUT = CALL_STATE_MACHINE_CALLOUT,
  49         ENGINE_STATE_CALLIN = CALL_STATE_MACHINE_CALLIN,
  50         ENGINE_STATE_SPEAKING = CALL_STATE_MACHINE_SPEAKING,
  51         __ENGINE_STATE_MAX,
  52 };
  53
  54 enum engine_sound_type {
  55         ENGINE_SOUND_DIAL = 0,
  56         ENGINE_SOUND_RING,
  57         ENGINE_SOUND_BUSY,
  58 };
  59
  60 struct engine_state {
  61         volatile enum engine_state_num state;
  62         enum engine_state_num (*process)(int, int *, int, int,
  63                                          struct alsa_dev *);
  64 };
  65
  66 #define STATE_MAP_SET(s, f)  {  \
  67         .state = (s),           \
  68         .process = (f)          \
  69 }
  70
  71 extern volatile sig_atomic_t quit;
  72
  73 extern void *engine_main(void *arg);
  74
  75 volatile sig_atomic_t stun_done = 0;
  76
  77 static char *alsadev = "plughw:0,0"; //XXX
  78 static char *port = "30111"; //XXX
  79
  80 struct engine_curr {
  81         int active;
  82         int sock;
  83         struct sockaddr addr;
  84         socklen_t addrlen;
  85 };
  86
  87 static struct engine_curr ecurr;
  88
  89 static void engine_play_file(struct alsa_dev *dev, enum engine_sound_type type)
  90 {
  91         int fd, nfds;
  92         char path[PATH_MAX];
  93         short pcm[FRAME_SIZE];
  94         struct pollfd *pfds = NULL;
  95
  96         memset(path, 0, sizeof(path));
  97         switch (type) {
  98         case ENGINE_SOUND_DIAL:
  99                 slprintf(path, sizeof(path), "%s/%s", FILE_ETCDIR, FILE_DIAL);
 100                 break;
 101         case ENGINE_SOUND_BUSY:
 102                 slprintf(path, sizeof(path), "%s/%s", FILE_ETCDIR, FILE_BUSY);
 103                 break;
 104         case ENGINE_SOUND_RING:
 105                 slprintf(path, sizeof(path), "%s/%s", FILE_ETCDIR, FILE_RING);
 106                 break;
 107         }
 108
 109         fd = open(path, O_RDONLY);
 110         if (fd < 0)
 111                 panic("Cannot open ring file!\n");
 112
 113         alsa_start(dev);
 114
 115         nfds = alsa_nfds(dev);
 116         pfds = xmalloc(sizeof(*pfds) * nfds);
 117
 118         alsa_getfds(dev, pfds, nfds);
 119
 120         memset(pcm, 0, sizeof(pcm));
 121         while (read(fd, pcm, sizeof(pcm)) > 0) {
 122                 poll(pfds, nfds, -1);
 123
 124                 alsa_write(dev, pcm, FRAME_SIZE);
 125                 memset(pcm, 0, sizeof(pcm));
 126
 127                 alsa_read(dev, pcm, FRAME_SIZE);
 128                 memset(pcm, 0, sizeof(pcm));
 129         }
 130
 131         alsa_stop(dev);
 132         close(fd);
 133         xfree(pfds);
 134 }
 135
 136 static inline void engine_play_ring(struct alsa_dev *dev)
 137 {
 138         engine_play_file(dev, ENGINE_SOUND_RING);
 139 }
 140
 141 static inline void engine_play_busy(struct alsa_dev *dev)
 142 {
 143         engine_play_file(dev, ENGINE_SOUND_BUSY);
 144 }
 145
 146 static inline void engine_play_dial(struct alsa_dev *dev)
 147 {
 148         engine_play_file(dev, ENGINE_SOUND_DIAL);
 149 }
 150
 151 static void engine_decode_packet(uint8_t *pkt, size_t len)
 152 {
 153         struct transsip_hdr *hdr;
 154
 155         if (len < sizeof(*hdr)) {
 156                 whine("[dbg] pkt too small!\n");
 157                 return;
 158         }
 159
 160         hdr = (struct transsip_hdr *) pkt;
 161         whine("[dbg] packet:\n");
 162         whine("[dbg]   seq: %d\n", hdr->seq);
 163         whine("[dbg]   est: %d\n", hdr->est);
 164         whine("[dbg]   psh: %d\n", hdr->psh);
 165         whine("[dbg]   bsy: %d\n", hdr->bsy);
 166         whine("[dbg]   fin: %d\n", hdr->fin);
 167         whine("[dbg]   res1: %d\n", hdr->res1);
 168 }
 169
 170 static enum engine_state_num engine_do_callout(int ssock, int *csock, int usocki,
 171                                                int usocko, struct alsa_dev *dev)
 172 {
 173         int one, mtu, tries, i;
 174         ssize_t ret;
 175         struct cli_pkt cpkt;
 176         struct addrinfo hints, *ahead, *ai;
 177         char msg[MAX_MSG];
 178         struct pollfd fds[2];
 179         struct sockaddr raddr;
 180         socklen_t raddrlen;
 181         struct transsip_hdr *thdr;
 182
 183         assert(ecurr.active == 0);
 184
 185         memset(&cpkt, 0, sizeof(cpkt));
 186         ret = read(usocki, &cpkt, sizeof(cpkt));
 187         if (ret != sizeof(cpkt)) {
 188                 whine("Read error from cli!\n");
 189                 return ENGINE_STATE_IDLE;
 190         }
 191         if (cpkt.ring == 0) {
 192                 whine("Read no ring flag from cli!\n");
 193                 return ENGINE_STATE_IDLE;
 194         }
 195
 196         memset(&hints, 0, sizeof(hints));
 197         hints.ai_family = PF_UNSPEC;
 198         hints.ai_socktype = SOCK_DGRAM;
 199         hints.ai_protocol = IPPROTO_UDP;
 200         hints.ai_flags = AI_NUMERICSERV;
 201
 202         ret = getaddrinfo(cpkt.address, cpkt.port, &hints, &ahead);
 203         if (ret < 0) {
 204                 whine("Cannot get address info for %s:%s!\n",
 205                       cpkt.address, cpkt.port);
 206                 return ENGINE_STATE_IDLE;
 207         }
 208
 209         *csock = -1;
 210
 211         for (ai = ahead; ai != NULL && *csock < 0; ai = ai->ai_next) {
 212                 *csock = socket(ai->ai_family, ai->ai_socktype,
 213                                 ai->ai_protocol);
 214                 if (*csock < 0)
 215                         continue;
 216
 217                 ret = connect(*csock, ai->ai_addr, ai->ai_addrlen);
 218                 if (ret < 0) {
 219                         whine("Cannot connect to remote!\n");
 220                         close(*csock);
 221                         *csock = -1;
 222                         continue;
 223                 }
 224
 225                 one = 1;
 226                 setsockopt(*csock, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
 227
 228                 mtu = IP_PMTUDISC_DONT;
 229                 setsockopt(*csock, SOL_IP, IP_MTU_DISCOVER, &mtu, sizeof(mtu));
 230
 231                 memcpy(&ecurr.addr, ai->ai_addr, ai->ai_addrlen);
 232                 ecurr.addrlen = ai->ai_addrlen;
 233                 ecurr.sock = *csock;
 234                 ecurr.active = 0;
 235         }
 236         freeaddrinfo(ahead);
 237
 238         if (*csock < 0) {
 239                 whine("Cannot connect to server!\n");
 240                 return ENGINE_STATE_IDLE;
 241         }
 242
 243         tries = 0;
 244
 245         memset(msg, 0, sizeof(msg));
 246         thdr = (struct transsip_hdr *) msg;
 247         thdr->est = 1;
 248
 249         ret = sendto(*csock, msg, sizeof(*thdr), 0, &ecurr.addr,
 250                      ecurr.addrlen);
 251         if (ret <= 0) {
 252                 whine("Cannot send ring probe to server!\n");
 253                 goto out_err;
 254         }
 255
 256         fds[0].fd = *csock;
 257         fds[0].events = POLLIN;
 258         fds[1].fd = usocki;
 259         fds[1].events = POLLIN;
 260
 261         while (!quit && tries++ < 100) {
 262                 poll(fds, array_size(fds), 1500);
 263
 264                 for (i = 0; i < array_size(fds); ++i) {
 265                         if ((fds[i].revents & POLLERR) == POLLERR) {
 266                                 printf("Destination unreachable?\n");
 267                                 goto out_err;
 268                         }
 269                         if ((fds[i].revents & POLLIN) != POLLIN)
 270                                 continue;
 271
 272                         if (fds[i].fd == usocki) {
 273                                 ret = read(usocki, &cpkt, sizeof(cpkt));
 274                                 if (ret <= 0) {
 275                                         whine("Read error from cli!\n");
 276                                         continue;
 277                                 }
 278                                 if (cpkt.fin) {
 279                                         memset(&msg, 0, sizeof(msg));
 280                                         thdr = (struct transsip_hdr *) msg;
 281                                         thdr->bsy = 1;
 282                                         thdr->fin = 1;
 283
 284                                         sendto(*csock, msg, sizeof(*thdr), 0,
 285                                                &ecurr.addr, ecurr.addrlen);
 286
 287                                         whine("You aborted call!\n");
 288                                         goto out_err;
 289                                 }
 290                         }
 291
 292                         if (fds[i].fd == *csock) {
 293                                 memset(msg, 0, sizeof(msg));
 294                                 raddrlen = sizeof(raddr);
 295                                 ret = recvfrom(*csock, msg, sizeof(msg), 0,
 296                                                &raddr, &raddrlen);
 297                                 if (ret <= 0)
 298                                         continue;
 299
 300                                 if (raddrlen != ecurr.addrlen)
 301                                         continue;
 302                                 if (memcmp(&raddr, &ecurr.addr, raddrlen))
 303                                         continue;
 304
 305                                 thdr = (struct transsip_hdr *) msg;
 306                                 if (thdr->est == 1 && thdr->psh == 1) {
 307                                         ecurr.active = 1;
 308                                         whine("Call established!\n");
 309                                         return ENGINE_STATE_SPEAKING;
 310                                 }
 311                                 if (thdr->bsy == 1 || thdr->fin == 1) {
 312                                         whine("Remote end busy!\n");
 313                                         engine_play_busy(dev);
 314                                         engine_play_busy(dev);
 315                                         goto out_err;
 316                                 }
 317                         }
 318                 }
 319
 320                 engine_play_dial(dev);
 321         }
 322
 323 out_err:
 324         close(*csock);
 325         *csock = 0;
 326         return ENGINE_STATE_IDLE;
 327 }
 328
 329 static enum engine_state_num engine_do_callin(int ssock, int *csock, int usocki,
 330                                               int usocko, struct alsa_dev *dev)
 331 {
 332         int i;
 333         ssize_t ret;
 334         char msg[MAX_MSG];
 335         struct sockaddr raddr;
 336         socklen_t raddrlen;
 337         struct transsip_hdr *thdr;
 338         char hbuff[256], sbuff[256];
 339         struct pollfd fds[2];
 340         struct cli_pkt cpkt;
 341
 342         assert(ecurr.active == 0);
 343
 344         memset(&msg, 0, sizeof(msg));
 345         raddrlen = sizeof(raddr);
 346         ret = recvfrom(ssock, msg, sizeof(msg), 0, &raddr, &raddrlen);
 347         if (ret <= 0) {
 348                 whine("Receive error %s!\n", strerror(errno));
 349                 return ENGINE_STATE_IDLE;
 350         }
 351
 352         thdr = (struct transsip_hdr *) msg;
 353         if (thdr->est != 1)
 354                 return ENGINE_STATE_IDLE;
 355
 356         memcpy(&ecurr.addr, &raddr, raddrlen);
 357         ecurr.addrlen = raddrlen;
 358         ecurr.sock = ssock;
 359         ecurr.active = 0;
 360
 361         memset(hbuff, 0, sizeof(hbuff));
 362         memset(sbuff, 0, sizeof(sbuff));
 363         getnameinfo((struct sockaddr *) &raddr, raddrlen, hbuff, sizeof(hbuff),
 364                     sbuff, sizeof(sbuff), NI_NUMERICHOST | NI_NUMERICSERV);
 365
 366         printf("New incoming connection from %s:%s!\n", hbuff, sbuff);
 367         printf("Answer it with: take\n");
 368         printf("Reject it with: hangup\n");
 369         fflush(stdout);
 370
 371         fds[0].fd = ssock;
 372         fds[0].events = POLLIN;
 373         fds[1].fd = usocki;
 374         fds[1].events = POLLIN;
 375
 376         while (likely(!quit)) {
 377                 poll(fds, array_size(fds), 1500);
 378
 379                 for (i = 0; i < array_size(fds); ++i) {
 380                         if ((fds[i].revents & POLLIN) != POLLIN)
 381                                 continue;
 382
 383                         if (fds[i].fd == ssock) {
 384                                 memset(msg, 0, sizeof(msg));
 385                                 raddrlen = sizeof(raddr);
 386                                 ret = recvfrom(ssock, msg, sizeof(msg), 0,
 387                                                &raddr, &raddrlen);
 388                                 if (ret <= 0)
 389                                         continue;
 390                                 if (raddrlen != ecurr.addrlen)
 391                                         continue;
 392                                 if (memcmp(&raddr, &ecurr.addr, raddrlen))
 393                                         continue;
 394
 395                                 if (thdr->fin == 1 || thdr->bsy == 1) {
 396                                         whine("Remote end hung up!\n");
 397                                         engine_play_busy(dev);
 398                                         engine_play_busy(dev);
 399                                         goto out_err;
 400                                 }
 401
 402                         }
 403
 404                         if (fds[i].fd == usocki) {
 405                                 ret = read(usocki, &cpkt, sizeof(cpkt));
 406                                 if (ret <= 0) {
 407                                         whine("Error reading from cli!\n");
 408                                         continue;
 409                                 }
 410                                 if (cpkt.fin) {
 411                                         memset(&msg, 0, sizeof(msg));
 412                                         thdr = (struct transsip_hdr *) msg;
 413                                         thdr->bsy = 1;
 414                                         thdr->fin = 1;
 415
 416                                         sendto(ssock, msg, sizeof(*thdr), 0,
 417                                                &ecurr.addr, ecurr.addrlen);
 418
 419                                         whine("You aborted call!\n");
 420                                         goto out_err;
 421                                 }
 422                                 if (cpkt.take) {
 423                                         memset(&msg, 0, sizeof(msg));
 424                                         thdr = (struct transsip_hdr *) msg;
 425                                         thdr->est = 1;
 426                                         thdr->psh = 1;
 427
 428                                         ret = sendto(ssock, msg, sizeof(*thdr), 0,
 429                                                      &ecurr.addr, ecurr.addrlen);
 430                                         if (ret <= 0) {
 431                                                 whine("Error sending ack!\n");
 432                                                 goto out_err;
 433                                         }
 434
 435                                         ecurr.active = 1;
 436                                         whine("Call established!\n");
 437                                         return ENGINE_STATE_SPEAKING;
 438                                 }
 439                         }
 440                 }
 441
 442                 engine_play_ring(dev);
 443         }
 444
 445 out_err:
 446         return ENGINE_STATE_IDLE;
 447 }
 448
 449 static enum engine_state_num engine_do_speaking(int ssock, int *csock,
 450                                                 int usocki, int usocko,
 451                                                 struct alsa_dev *dev)
 452 {
 453         ssize_t ret;
 454         int recv_started = 0, nfds = 0, tmp, i, one;
 455         struct pollfd *pfds = NULL;
 456         char msg[MAX_MSG];
 457         uint32_t send_seq = 0;
 458         CELTMode *mode;
 459         CELTEncoder *encoder;
 460         CELTDecoder *decoder;
 461         JitterBuffer *jitter;
 462         SpeexPreprocessState *preprocess;
 463         SpeexEchoState *echo_state;
 464         struct sockaddr raddr;
 465         struct transsip_hdr *thdr;
 466         socklen_t raddrlen;
 467         struct cli_pkt cpkt;
 468
 469         assert(ecurr.active == 1);
 470
 471         mode = celt_mode_create(SAMPLING_RATE, FRAME_SIZE, NULL);
 472         encoder = celt_encoder_create(mode, 1, NULL);
 473         decoder = celt_decoder_create(mode, 1, NULL);
 474
 475         jitter = jitter_buffer_init(FRAME_SIZE);
 476         tmp = FRAME_SIZE;
 477         jitter_buffer_ctl(jitter, JITTER_BUFFER_SET_MARGIN, &tmp);
 478
 479         echo_state = speex_echo_state_init(FRAME_SIZE, 10 * FRAME_SIZE);
 480         tmp = SAMPLING_RATE;
 481         speex_echo_ctl(echo_state, SPEEX_ECHO_SET_SAMPLING_RATE, &tmp);
 482
 483         one = 1;
 484         preprocess = speex_preprocess_state_init(FRAME_SIZE, SAMPLING_RATE);
 485         speex_preprocess_ctl(preprocess, SPEEX_PREPROCESS_SET_DENOISE, &one);
 486         speex_preprocess_ctl(preprocess, SPEEX_PREPROCESS_SET_AGC, &one);
 487         speex_preprocess_ctl(preprocess, SPEEX_PREPROCESS_SET_DEREVERB, &one);
 488         speex_preprocess_ctl(preprocess, SPEEX_PREPROCESS_SET_ECHO_STATE,
 489                              echo_state);
 490
 491         nfds = alsa_nfds(dev);
 492         pfds = xmalloc(sizeof(*pfds) * (nfds + 2));
 493
 494         alsa_getfds(dev, pfds, nfds);
 495
 496         pfds[nfds].fd = ecurr.sock;
 497         pfds[nfds].events = POLLIN;
 498         pfds[nfds + 1].fd = usocki;
 499         pfds[nfds + 1].events = POLLIN;
 500
 501         alsa_start(dev);
 502
 503         while (likely(!quit)) {
 504                 poll(pfds, nfds + 2, -1);
 505
 506                 if (pfds[nfds + 1].revents & POLLIN) {
 507                         ret = read(usocki, &cpkt, sizeof(cpkt));
 508                         if (ret <= 0) {
 509                                 whine("Read error from cli!\n");
 510                                 continue;
 511                         }
 512                         if (cpkt.fin) {
 513                                 whine("You aborted call!\n");
 514                                 goto out_err;
 515                         }
 516                 }
 517
 518                 if (pfds[nfds].revents & POLLIN) {
 519                         JitterBufferPacket packet;
 520
 521                         memset(msg, 0, sizeof(msg));
 522                         raddrlen = sizeof(raddr);
 523                         ret = recvfrom(ecurr.sock, msg, sizeof(msg), 0,
 524                                        &raddr, &raddrlen);
 525                         if (unlikely(ret <= 0))
 526                                 continue;
 527
 528                         if (raddrlen != ecurr.addrlen ||
 529                             memcmp(&raddr, &ecurr.addr, raddrlen)) {
 530                                 memset(msg, 0, sizeof(msg));
 531
 532                                 thdr = (struct transsip_hdr *) msg;
 533                                 thdr->bsy = 1;
 534                                 thdr->fin = 1;
 535
 536                                 sendto(ecurr.sock, msg, sizeof(*thdr), 0,
 537                                        &raddr, raddrlen);
 538
 539                                 goto out_alsa;
 540                         }
 541
 542                         thdr = (struct transsip_hdr *) msg;
 543                         if (thdr->fin == 1) {
 544                                 whine("Remote end hung up!\n");
 545                                 goto out_err;
 546                         }
 547
 548                         packet.data = msg + sizeof(*thdr);
 549                         packet.len = ret - sizeof(*thdr);
 550                         packet.timestamp = ntohl(thdr->seq);
 551                         packet.span = FRAME_SIZE;
 552                         packet.sequence = 0;
 553
 554                         jitter_buffer_put(jitter, &packet);
 555                         recv_started = 1;
 556                 }
 557 out_alsa:
 558                 if (alsa_play_ready(dev, pfds, nfds)) {
 559                         short pcm[FRAME_SIZE];
 560
 561                         if (recv_started) {
 562                                 JitterBufferPacket packet;
 563
 564                                 memset(msg, 0, sizeof(msg));
 565                                 packet.data = msg;
 566                                 packet.len  = MAX_MSG;
 567
 568                                 jitter_buffer_tick(jitter);
 569                                 jitter_buffer_get(jitter, &packet,
 570                                                   FRAME_SIZE, NULL);
 571                                 if (packet.len == 0)
 572                                         packet.data = NULL;
 573
 574                                 celt_decode(decoder, (const unsigned char *)
 575                                             packet.data, packet.len, pcm);
 576                         } else {
 577                                 for (i = 0; i < FRAME_SIZE; ++i)
 578                                         pcm[i] = 0;
 579                         }
 580
 581                         if (alsa_write(dev, pcm, FRAME_SIZE))
 582                                 speex_echo_state_reset(echo_state);
 583                         speex_echo_playback(echo_state, pcm);
 584                 }
 585
 586                 if (alsa_cap_ready(dev, pfds, nfds)) {
 587                         short pcm[FRAME_SIZE];
 588                         short pcm2[FRAME_SIZE];
 589
 590                         memset(msg, 0, sizeof(msg));
 591
 592                         alsa_read(dev, pcm, FRAME_SIZE);
 593
 594                         speex_echo_capture(echo_state, pcm, pcm2);
 595                         for (i = 0; i < FRAME_SIZE; ++i)
 596                                 pcm[i] = pcm2[i];
 597
 598                         speex_preprocess_run(preprocess, pcm);
 599
 600                         celt_encode(encoder, pcm, NULL, (unsigned char *)
 601                                     (msg + sizeof(*thdr)), PACKETSIZE);
 602
 603                         thdr = (struct transsip_hdr *) msg;
 604                         thdr->psh = 1;
 605                         thdr->est = 1;
 606                         thdr->seq = htonl(send_seq);
 607                         send_seq += FRAME_SIZE;
 608
 609                         ret = sendto(ecurr.sock, msg,
 610                                      PACKETSIZE + sizeof(*thdr), 0,
 611                                      &ecurr.addr, ecurr.addrlen);
 612                         if (ret <= 0) {
 613                                 whine("Send datagram failed!\n");
 614                                 goto out_err;
 615                         }
 616                 }
 617         }
 618
 619 out_err:
 620         alsa_stop(dev);
 621
 622         memset(msg, 0, sizeof(msg));
 623         thdr = (struct transsip_hdr *) msg;
 624         thdr->fin = 1;
 625
 626         sendto(ecurr.sock, msg, sizeof(*thdr), 0, &ecurr.addr,
 627                ecurr.addrlen);
 628
 629         if (ecurr.sock == *csock) {
 630                 close(*csock);
 631                 *csock = 0;
 632         }
 633
 634         celt_encoder_destroy(encoder);
 635         celt_decoder_destroy(decoder);
 636         celt_mode_destroy(mode);
 637
 638         speex_preprocess_state_destroy(preprocess);
 639         jitter_buffer_destroy(jitter);
 640
 641         xfree(pfds);
 642
 643         ecurr.active = 0;
 644         return ENGINE_STATE_IDLE;
 645 }
 646
 647 static inline void engine_drop_from_queue(int sock)
 648 {
 649         char msg[MAX_MSG];
 650         recv(sock, msg, sizeof(msg), 0);
 651 }
 652
 653 static enum engine_state_num engine_do_idle(int ssock, int *csock, int usocki,
 654                                             int usocko, struct alsa_dev *dev)
 655 {
 656         int i;
 657         ssize_t ret;
 658         struct pollfd fds[2];
 659         char msg[MAX_MSG];
 660         struct transsip_hdr *thdr;
 661         struct cli_pkt cpkt;
 662
 663         assert(ecurr.active == 0);
 664
 665         fds[0].fd = ssock;
 666         fds[0].events = POLLIN;
 667         fds[1].fd = usocki;
 668         fds[1].events = POLLIN;
 669
 670         while (likely(!quit)) {
 671                 memset(msg, 0, sizeof(msg));
 672
 673                 poll(fds, array_size(fds), 1000);
 674
 675                 for (i = 0; i < array_size(fds); ++i) {
 676                         if ((fds[i].revents & POLLIN) != POLLIN)
 677                                 continue;
 678
 679                         if (fds[i].fd == ssock) {
 680                                 ret = recv(ssock, msg, sizeof(msg), MSG_PEEK);
 681                                 if (ret < 0)
 682                                         continue;
 683                                 if (ret < sizeof(struct transsip_hdr))
 684                                         engine_drop_from_queue(ssock);
 685
 686                                 thdr = (struct transsip_hdr *) msg;
 687                                 if (thdr->est == 1 && thdr->psh == 0)
 688                                         return ENGINE_STATE_CALLIN;
 689                                 else
 690                                         engine_drop_from_queue(ssock);
 691                         }
 692
 693                         if (fds[i].fd == usocki) {
 694                                 ret = read(usocki, &cpkt, sizeof(cpkt));
 695                                 if (ret <= 0) {
 696                                         whine("Error reading from cli!\n");
 697                                         continue;
 698                                 }
 699                                 if (cpkt.ring)
 700                                         return ENGINE_STATE_CALLOUT;
 701                         }
 702                 }
 703         }
 704
 705         return ENGINE_STATE_IDLE;
 706 }
 707
 708 static struct engine_state state_machine[__ENGINE_STATE_MAX] __read_mostly = {
 709         STATE_MAP_SET(ENGINE_STATE_IDLE, engine_do_idle),
 710         STATE_MAP_SET(ENGINE_STATE_CALLOUT, engine_do_callout),
 711         STATE_MAP_SET(ENGINE_STATE_CALLIN, engine_do_callin),
 712         STATE_MAP_SET(ENGINE_STATE_SPEAKING, engine_do_speaking),
 713 };
 714
 715 void *engine_main(void *arg)
 716 {
 717         int ssock = -1, csock, ret, mtu, usocki, usocko;
 718         enum engine_state_num state;
 719         struct addrinfo hints, *ahead, *ai;
 720         struct alsa_dev *dev = NULL;
 721         struct pipepair *pp = arg;
 722
 723         init_call_notifier();
 724
 725         usocki = pp->i;
 726         usocko = pp->o;
 727
 728         while (!stun_done)
 729                 sleep(0);
 730
 731         memset(&hints, 0, sizeof(hints));
 732         hints.ai_family = PF_UNSPEC;
 733         hints.ai_socktype = SOCK_DGRAM;
 734         hints.ai_protocol = IPPROTO_UDP;
 735         hints.ai_flags = AI_PASSIVE;
 736
 737         ret = getaddrinfo(NULL, port, &hints, &ahead);
 738         if (ret < 0)
 739                 panic("Cannot get address info!\n");
 740
 741         for (ai = ahead; ai != NULL && ssock < 0; ai = ai->ai_next) {
 742                 ssock = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
 743                 if (ssock < 0)
 744                         continue;
 745                 if (ai->ai_family == AF_INET6) {
 746                         int one = 1;
 747 #ifdef IPV6_V6ONLY
 748                         ret = setsockopt(ssock, IPPROTO_IPV6, IPV6_V6ONLY,
 749                                          &one, sizeof(one));
 750                         if (ret < 0) {
 751                                 close(ssock);
 752                                 ssock = -1;
 753                                 continue;
 754                         }
 755 #else
 756                         close(ssock);
 757                         ssock = -1;
 758                         continue;
 759 #endif /* IPV6_V6ONLY */
 760                 }
 761
 762                 mtu = IP_PMTUDISC_DONT;
 763                 setsockopt(ssock, SOL_IP, IP_MTU_DISCOVER, &mtu, sizeof(mtu));
 764
 765                 ret = bind(ssock, ai->ai_addr, ai->ai_addrlen);
 766                 if (ret < 0) {
 767                         close(ssock);
 768                         ssock = -1;
 769                         continue;
 770                 }
 771         }
 772
 773         freeaddrinfo(ahead);
 774         if (ssock < 0)
 775                 panic("Cannot open socket!\n");
 776
 777         dev = alsa_open(alsadev, SAMPLING_RATE, 1, FRAME_SIZE);
 778         if (!dev)
 779                 panic("Cannot open ALSA device %s!\n", alsadev);
 780
 781         state = ENGINE_STATE_IDLE;
 782         while (likely(!quit)) {
 783                 int arg = state;
 784                 call_notifier_exec(CALL_STATE_MACHINE_CHANGED, &arg);
 785                 state = state_machine[state].process(ssock, &csock,
 786                                                      usocki, usocko,
 787                                                      dev);
 788         }
 789
 790         alsa_close(dev);
 791         close(ssock);
 792
 793         pthread_exit(0);
 794 }