su_proxy.c

this is simple sip stack.

  
  if (su_wait_create(f->f_wait, f->f_socket, SU_WAIT_ACCEPT) < 0)
    return SOCKET_ERROR;

  if (su_root_register(f->f_pr->pr_root, f->f_wait, 
		       forwarder_accept, f, 0) < 0)
    return SOCKET_ERROR;

  return 0;
}

int forwarder_init_dgram(forwarder_t *f)
{
  /* Unimplemented */
  return SOCKET_ERROR;
}

/** Accept a connection. */
int forwarder_accept(proxy_t *pr, su_wait_t *w, forwarder_t *f0)
{
  forwarder_t *f;
  su_sockaddr_t *su;
  socklen_t  sulen;
  int events;
  
  events = su_wait_events(w, f0->f_socket);

  f = forwarder_create(pr);

  if (f) {
    su = f->f_dest;
    sulen = sizeof(f->f_dest);
    f->f_socket = accept(f0->f_socket, &su->su_sa, &sulen);
    f->f_upstream = 1;
    if (f->f_socket != SOCKET_ERROR) {
      char buf[SU_ADDRSIZE];
      
      SU_DEBUG_3(("accept: connection from %s:%u\n",
		  inet_ntop(su->su_family, SU_ADDR(su), buf, sizeof(buf)),
		  ntohs(su->su_port)));

      if (!su_wait_create(f->f_wait, f->f_socket, SU_WAIT_IN) &&
	  !su_wait_create(f->f_wait + 1, f->f_socket, SU_WAIT_OUT)) {
	if (forwarder_stream_peer(pr, f) != SOCKET_ERROR) {
	  /* success */
	  return 0;
	}
      }
      else {
	SU_DEBUG_1(("%s: cannot create wait objects\n", __func__));
      }
    }
  }

  forwarder_destroy(f);

  return 0;
}

int forwarder_stream_peer(proxy_t *pr, forwarder_t *f_peer)
{
  forwarder_t *f;
  su_addrinfo_t *ai;

  assert(f_peer);
  
  f = forwarder_create(pr);
  if (!f) {
    SU_DEBUG_1(("%s: cannot allocate peer\n", __func__));
    goto error;
  }

  for (ai = pr->pr_addrinfo; ai; ai = ai->ai_next) {
    if (ai->ai_socktype == SOCK_STREAM)
      break;
  }

  if (!ai) {
    SU_DEBUG_1(("%s: no matching destination\n", __func__));
    goto error;
  }

  f->f_socket = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
  if (f->f_socket == INVALID_SOCKET) {
    SU_DEBUG_1(("%s: socket: %s\n", __func__, su_strerror(su_errno())));
    goto error;
  }

  if (su_wait_create(f->f_wait, f->f_socket, SU_WAIT_IN) ||
      su_wait_create(f->f_wait + 1, f->f_socket, SU_WAIT_OUT)) {
    SU_DEBUG_1(("%s: cannot create wait objects\n", __func__));
    goto error;
  }

  /* Asynchronous connect */
  su_setblocking(f->f_socket, 0);
  su_seterrno(0);
  connect(f->f_socket, ai->ai_addr, ai->ai_addrlen);
  memcpy(f->f_dest, ai->ai_addr, ai->ai_addrlen);

  if (su_errno() != EINPROGRESS) {
    SU_DEBUG_1(("%s: connect: %s\n", __func__, su_strerror(su_errno())));
    goto error;
  }

  if (su_root_register(pr->pr_root, f->f_wait + 1, 
		       forwarder_connected, f, 0) == SOCKET_ERROR) {
    SU_DEBUG_1(("%s: cannot register\n", __func__));
    goto error;
  }

  f->f_peer = f_peer;
  f_peer->f_peer = f;
  return 0;

 error:
  forwarder_destroy(f);
  return SOCKET_ERROR;
}

/** Connection is complete. */
int forwarder_connected(proxy_t *pr, su_wait_t *w, forwarder_t *f)
{
  int events, error;
  forwarder_t *f_peer;

  events = su_wait_events(w, f->f_socket);

  error = su_soerror(f->f_socket);

  if (error) {
    SU_DEBUG_1(("connect: %s\n", su_strerror(error)));
    forwarder_destroy(f);
    return 0;
  }

  su_root_unregister(pr->pr_root, f->f_wait + 1, forwarder_connected, f);

  /* Wait for data, forward it to peer */
  assert(f->f_peer);
  f_peer = f->f_peer;
  su_root_register(pr->pr_root, f->f_wait, forwarder_recv, f, 0);
  su_root_register(pr->pr_root, f_peer->f_wait, forwarder_recv, f_peer, 0);

  return 0;
}

/** Receive data, forward it to peer */
int forwarder_recv(proxy_t *pr, su_wait_t *w, forwarder_t *f)
{
  buffer_t b[1];
  int n, events;

  events = su_wait_events(w, f->f_socket);

  n = recv(f->f_socket, b->b_data, sizeof(b->b_data), 0);

  if (n > 0) {
    b->b_sent = 0; b->b_used = n;
    if (f->f_peer->f_buf) {
      forwarder_append(f, b); 
      return 0;
    }
    if (forwarder_send(pr, f->f_peer, b) >= 0) {
      if (b->b_sent < b->b_used) {
	su_root_unregister(pr->pr_root, w, forwarder_recv, f);
	su_root_register(pr->pr_root, f->f_peer->f_wait + 1, 
			 forwarder_empty, f->f_peer, 0);
	forwarder_append(f, b);
      }
      return 0;
    }
    else {
      /* Error when sending */
    }
  }
  if (n < 0) {
    int error = su_errno();
    SU_DEBUG_1(("recv: %s\n", su_strerror(error)));
  
    if (error == EINTR || error == EAGAIN || error == EWOULDBLOCK) {
      return 0;
    }
    /* XXX */
    forwarder_destroy(f);
  }

  /* shutdown */
  forwarder_shutdown(f);	

  return 0;
}

int forwarder_send(proxy_t *pr, forwarder_t *f, buffer_t *b)
{
  int n, error;
  
  do {
    n = send(f->f_socket, b->b_data + b->b_sent, b->b_used - b->b_sent, 0);

    if (n < 0) {
      error = su_errno();
      if (error == EINTR)
	continue;
      SU_DEBUG_1(("send: %s\n", su_strerror(error)));
      if (error != EAGAIN && error != EWOULDBLOCK)
	return -error;
    }
    else {
      f->f_sent += n;
    }
  }
  while (n > 0 && (b->b_sent += n) < b->b_used);

  return b->b_used - b->b_sent;
}

int forwarder_append(forwarder_t *f, buffer_t *b0)
{
  buffer_t *b, **bb;
  int unsent;

  /* Find last buffer */
  for (bb = &f->f_buf; *bb; bb = &(*bb)->b_next)
    ;

  unsent = b0->b_used - b0->b_sent;
  assert(unsent > 0);

  b = su_alloc(f->f_pr->pr_home, offsetof(buffer_t, b_data[unsent]));
  if (b) {
    *bb = b;
    b->b_next = NULL;
    b->b_prev = bb;
    b->b_used = unsent;
    b->b_sent = 0;
    memcpy(b->b_data, b0->b_data + b->b_sent, unsent);
  }

  return b ? 0 : -1;
}

/** Empty forwarder buffers */
int forwarder_empty(proxy_t *pr, su_wait_t *w, forwarder_t *f)
{
  buffer_t *b;
  int n, events;

  events = su_wait_events(w, f->f_socket);

  while ((b = f->f_buf)) {
    n = forwarder_send(f->f_pr, f, b);
    if (n == 0) {
      if ((f->f_buf = b->b_next))
	b->b_next->b_prev = &f->f_buf;
      su_free(f->f_pr->pr_home, b);
      continue;
    }
    else if (n < 0) {
      /* XXX */
    }
    break;
  }

  if (!f->f_buf) {
    forwarder_t *f_peer = f->f_peer;

    su_root_unregister(pr->pr_root, w, forwarder_empty, f);

    if (!f->f_shutdown) {
      /* Buffer is empty - start receiving */
      su_root_register(pr->pr_root, f_peer->f_wait, forwarder_recv, f_peer, 0);
    }
    else {
      if (shutdown(f->f_socket, 1) < 0) {
	SU_DEBUG_1(("shutdown(1): %s\n", su_strerror(su_errno())));
      }
      if (f_peer->f_shutdown) {
	forwarder_close(f);
      }
    }
  }

  return 0;
}

int forwarder_shutdown(forwarder_t *f)
{
  forwarder_t *f_peer = f->f_peer;
  su_sockaddr_t *su = f->f_dest;
  char buf[SU_ADDRSIZE];
  
  SU_DEBUG_3(("forwarder_shutdown: shutdown from %s:%u\n",
	      inet_ntop(su->su_family, SU_ADDR(su), buf, sizeof(buf)),
	      ntohs(su->su_port)));
  
  if (su_root_unregister(f->f_pr->pr_root, f->f_wait, forwarder_recv, f) < 0) {
    SU_DEBUG_1(("%s: su_root_unregister failed\n", __func__));
  }

  if (shutdown(f->f_socket, 0) < 0) {
    SU_DEBUG_1(("shutdown(0): %s\n", su_strerror(su_errno())));
  }
  f_peer->f_shutdown = 1;
  if (!f_peer->f_buf) {
    if (shutdown(f_peer->f_socket, 1) < 0) {
      SU_DEBUG_1(("shutdown(1): %s\n", su_strerror(su_errno())));
    }
    if (f->f_shutdown) {
      forwarder_close(f);
    }
  }
  return 0;
}

/** Close a peer pair */
void forwarder_close(forwarder_t *f)
{
  su_sockaddr_t *su1, *su2;
  char const *d1, *d2;
  char buf1[SU_ADDRSIZE], buf2[SU_ADDRSIZE];

  if (f->f_upstream)
    su1 = f->f_dest, su2 = f->f_peer->f_dest, d1 = "up", d2 = "down";
  else
    su2 = f->f_dest, su1 = f->f_peer->f_dest, d2 = "up", d1 = "down";

  SU_DEBUG_3(("forwarder_close: connection from %s:%u to %s:%d\n",
	      inet_ntop(su1->su_family, SU_ADDR(su1), buf1, sizeof(buf1)),
	      ntohs(su1->su_port),
	      inet_ntop(su2->su_family, SU_ADDR(su2), buf2, sizeof(buf2)),
	      ntohs(su2->su_port)));

  forwarder_destroy(f);
}