tport.c

Sofia SIP is an open-source SIP User-Agent library, compliant with the IETF RFC3261 specification.

    if (!su_is_blocking(err))
      TPORT_CONNECT_ERROR(err, connect);
    events = SU_WAIT_CONNECT | SU_WAIT_ERR;
    wakeup = tport_connected;
    what = "connecting";
  }
  else {
    what = "connected";
    self->tp_is_connected = 1;
  }

  if (su_wait_create(wait, s, self->tp_events = events) == -1)
    TPORT_CONNECT_ERROR(su_errno(), su_wait_create);

  /* Register receiving function with events specified above */
  if ((index = su_root_register(mr->mr_root, wait, wakeup, self, 0)) == -1)
    TPORT_CONNECT_ERROR(su_errno(), su_root_register);

  self->tp_index = index;

  if (ai == real_ai) {
    SU_DEBUG_5(("%s(%p): %s to " TPN_FORMAT "\n", 
		__func__, (void *)self, what, TPN_ARGS(self->tp_name)));
  }
  else {
    SU_DEBUG_5(("%s(%p): %s via %s to " TPN_FORMAT "\n",
		__func__, (void *)self, what,
		tport_hostport(buf, sizeof(buf), (void *)ai->ai_addr, 2),
		TPN_ARGS(self->tp_name)));
  }

  tprb_append(&pri->pri_open, self);

  return self;
}

/** Destroy a secondary transport. @internal */
void tport_zap_secondary(tport_t *self)
{
  tport_master_t *mr;

  if (self == NULL)
    return;

  /* Remove from rbtree */
  if (!tport_is_closed(self))
    tprb_remove(&self->tp_pri->pri_open, self);
  else
    tplist_remove(&self->tp_pri->pri_closed, self);

  if (self->tp_timer)
    su_timer_destroy(self->tp_timer), self->tp_timer = NULL;

  /* Do not deinit primary as secondary! */
  if (tport_is_secondary(self) &&
      self->tp_pri->pri_vtable->vtp_deinit_secondary)
    self->tp_pri->pri_vtable->vtp_deinit_secondary(self);

  if (self->tp_msg) {
    msg_destroy(self->tp_msg), self->tp_msg = NULL;
    SU_DEBUG_3(("%s(%p): zapped partially received message\n", 
		__func__, (void *)self));
  }

  if (tport_has_queued(self)) {
    size_t n = 0, i, N = self->tp_params->tpp_qsize;
    for (i = self->tp_qhead; self->tp_queue[i]; i = (i + 1) % N) {
      msg_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
      n++;
    }
    SU_DEBUG_3(("%s(%p): zapped %lu queued messages\n", 
		__func__, (void *)self, (LU)n));
  }

  if (self->tp_pused) {
    SU_DEBUG_3(("%s(%p): zapped while pending\n",
		__func__, (void *)self));
  }

  mr = self->tp_master;

#if HAVE_SOFIA_STUN
  tport_stun_server_remove_socket(self);
#endif

  if (self->tp_index)
    su_root_deregister(mr->mr_root, self->tp_index);
  self->tp_index = 0;
  if (self->tp_socket != INVALID_SOCKET)
    su_close(self->tp_socket);
  self->tp_socket = INVALID_SOCKET;

  su_home_zap(self->tp_home);
}

/** Create a new reference to a transport object. */
tport_t *tport_ref(tport_t *tp)
{
  if (tp) {
    if (tp->tp_refs >= 0)
      tp->tp_refs++;
    else if (tp->tp_refs == -1)
      tp->tp_refs = 1;
  }
  return tp;
}

/** Destroy reference to a transport object. */
void tport_unref(tport_t *tp)
{
  if (tp == NULL || tp->tp_refs <= 0)
    return;
  if (--tp->tp_refs > 0)
    return;
  
  if (!tport_is_secondary(tp))
    return;

  if (tp->tp_params->tpp_idle == 0)
    tport_close(tp);

  tport_set_secondary_timer(tp);
}

/** Create a new reference to transport object. */
tport_t *tport_incref(tport_t *tp)
{
  return tport_ref(tp);
}

/** Destroy a transport reference. */
void tport_decref(tport_t **ttp)
{
  assert(ttp);

  if (*ttp) {
    tport_unref(*ttp);
    *ttp = NULL;
  }
}

/** Get transport parameters.
 *
 * @param self          pointer to a transport object
 * @param tag,value,... list of tags
 *
 * @TAGS
 * TPTAG_MTU_REF(), TPTAG_QUEUESIZE_REF(), TPTAG_IDLE_REF(),
 * TPTAG_TIMEOUT_REF(), TPTAG_KEEPALIVE_REF(), TPTAG_PINGPONG_REF(),
 * TPTAG_PONG2PING_REF(), TPTAG_DEBUG_DROP_REF(), TPTAG_THRPSIZE_REF(),
 * TPTAG_THRPRQSIZE_REF(), TPTAG_SIGCOMP_LIFETIME_REF(),
 * TPTAG_CONNECT_REF(), TPTAG_SDWN_ERROR_REF(), TPTAG_REUSE_REF(),
 * TPTAG_STUN_SERVER_REF(), TPTAG_PUBLIC_REF() and TPTAG_TOS_REF().
 */
int tport_get_params(tport_t const *self,
		     tag_type_t tag, tag_value_t value, ...)
{
  ta_list ta;
  int n;
  tport_params_t const *tpp;
  int connect;

  if (self == NULL)
    return su_seterrno(EINVAL);

  tpp = self->tp_params;
  ta_start(ta, tag, value);

  connect = tpp->tpp_conn_orient 
    /* Only dgram primary is *not* connection-oriented */
    || !tport_is_primary(self) || !tport_is_dgram(self);

  n = tl_tgets(ta_args(ta),
	       TPTAG_MTU((usize_t)tpp->tpp_mtu),
	       TPTAG_REUSE(self->tp_reusable),
	       TPTAG_CONNECT(connect),
	       TPTAG_QUEUESIZE(tpp->tpp_qsize),
	       TPTAG_IDLE(tpp->tpp_idle),
	       TPTAG_TIMEOUT(tpp->tpp_timeout),
	       TPTAG_KEEPALIVE(tpp->tpp_keepalive),
	       TPTAG_PINGPONG(tpp->tpp_pingpong),
	       TPTAG_PONG2PING(tpp->tpp_pong2ping),
	       TPTAG_SDWN_ERROR(tpp->tpp_sdwn_error),
	       TPTAG_DEBUG_DROP(tpp->tpp_drop),
	       TPTAG_THRPSIZE(tpp->tpp_thrpsize),
	       TPTAG_THRPRQSIZE(tpp->tpp_thrprqsize),
	       TPTAG_SIGCOMP_LIFETIME(tpp->tpp_sigcomp_lifetime),
	       TPTAG_STUN_SERVER(tpp->tpp_stun_server),
	       TAG_IF(self->tp_pri,
		      TPTAG_PUBLIC(self->tp_pri ?
				   self->tp_pri->pri_public : 0)),
	       TPTAG_TOS(tpp->tpp_tos),
	       TAG_END());

  ta_end(ta);

  return n;
}

/** Set transport parameters.
 *
 * @param self          pointer to a transport object
 * @param tag,value,... list of tags
 *
 * @TAGS
 * TPTAG_MTU(), TPTAG_QUEUESIZE(), TPTAG_IDLE(), TPTAG_TIMEOUT(),
 * TPTAG_KEEPALIVE(), TPTAG_PINGPONG(), TPTAG_PONG2PING(),
 * TPTAG_DEBUG_DROP(), TPTAG_THRPSIZE(), TPTAG_THRPRQSIZE(),
 * TPTAG_SIGCOMP_LIFETIME(), TPTAG_CONNECT(), TPTAG_SDWN_ERROR(),
 * TPTAG_REUSE(), TPTAG_STUN_SERVER(), and TPTAG_TOS().
 */
int tport_set_params(tport_t *self,
		     tag_type_t tag, tag_value_t value, ...)
{
  ta_list ta;
  int n, m = 0;
  tport_params_t tpp[1], *tpp0;
  
  usize_t mtu;
  int connect, sdwn_error, reusable, stun_server, pong2ping;

  if (self == NULL)
    return su_seterrno(EINVAL);

  memcpy(tpp, tpp0 = self->tp_params, sizeof tpp);

  mtu = tpp->tpp_mtu;
  connect = tpp->tpp_conn_orient;
  sdwn_error = tpp->tpp_sdwn_error;
  reusable = self->tp_reusable;
  stun_server = tpp->tpp_stun_server;
  pong2ping = tpp->tpp_pong2ping;

  ta_start(ta, tag, value);

  n = tl_gets(ta_args(ta),
	      TPTAG_MTU_REF(mtu),
	      TAG_IF(!self->tp_queue, TPTAG_QUEUESIZE_REF(tpp->tpp_qsize)),
	      TPTAG_IDLE_REF(tpp->tpp_idle),
	      TPTAG_TIMEOUT_REF(tpp->tpp_timeout),
	      TPTAG_KEEPALIVE_REF(tpp->tpp_keepalive),
	      TPTAG_PINGPONG_REF(tpp->tpp_pingpong),
	      TPTAG_PONG2PING_REF(pong2ping),
	      TPTAG_DEBUG_DROP_REF(tpp->tpp_drop),
	      TPTAG_THRPSIZE_REF(tpp->tpp_thrpsize),
	      TPTAG_THRPRQSIZE_REF(tpp->tpp_thrprqsize),
	      TPTAG_SIGCOMP_LIFETIME_REF(tpp->tpp_sigcomp_lifetime),
	      TPTAG_CONNECT_REF(connect),
	      TPTAG_SDWN_ERROR_REF(sdwn_error),
	      TPTAG_REUSE_REF(reusable),
	      TPTAG_STUN_SERVER_REF(stun_server),
	      TPTAG_TOS_REF(tpp->tpp_tos),
	      TAG_END());

  if (self == (tport_t *)self->tp_master)
    m = tport_open_log(self->tp_master, ta_args(ta));

  ta_end(ta);

  if (n == 0)
    return m;

  if (tpp->tpp_idle > 0 && tpp->tpp_idle < 100)
    tpp->tpp_idle = 100;
  if (tpp->tpp_timeout < 100)
    tpp->tpp_timeout = 100;
  if (tpp->tpp_drop > 1000)
    tpp->tpp_drop = 1000;
  if (tpp->tpp_thrprqsize > 0)
    tpp->tpp_thrprqsize = tpp0->tpp_thrprqsize;
  if (tpp->tpp_sigcomp_lifetime != 0 && tpp->tpp_sigcomp_lifetime < 30)
    tpp->tpp_sigcomp_lifetime = 30;
  if (tpp->tpp_qsize >= 1000)
    tpp->tpp_qsize = 1000;

  if (mtu > UINT_MAX)
    mtu = UINT_MAX;
  tpp->tpp_mtu = (unsigned)mtu;
  /* Currently only primary UDP transport can *not* be connection oriented */ 
  tpp->tpp_conn_orient = connect; 
  tpp->tpp_sdwn_error = sdwn_error;
  self->tp_reusable = reusable;
  tpp->tpp_stun_server = stun_server;
  tpp->tpp_pong2ping = pong2ping;

  if (memcmp(tpp0, tpp, sizeof tpp) == 0)
    return n;

  if (tport_is_secondary(self) && 
      self->tp_params == self->tp_pri->pri_primary->tp_params) {
    tpp0 = su_zalloc(self->tp_home, sizeof *tpp0); if (!tpp0) return -1;
    self->tp_params = tpp0;
  }

  memcpy(tpp0, tpp, sizeof tpp);

  if (tport_is_secondary(self))
    tport_set_secondary_timer(self);

  return n + m;
}

extern tport_vtable_t const tport_udp_vtable;
extern tport_vtable_t const tport_tcp_vtable;
extern tport_vtable_t const tport_tls_vtable;
extern tport_vtable_t const tport_sctp_vtable;
extern tport_vtable_t const tport_udp_client_vtable;
extern tport_vtable_t const tport_tcp_client_vtable;
extern tport_vtable_t const tport_sctp_client_vtable;
extern tport_vtable_t const tport_tls_client_vtable;
extern tport_vtable_t const tport_http_connect_vtable;
extern tport_vtable_t const tport_threadpool_vtable;

#define TPORT_NUMBER_OF_TYPES 64

tport_vtable_t const *tport_vtables[TPORT_NUMBER_OF_TYPES + 1] =
{
#if HAVE_SOFIA_NTH
  &tport_http_connect_vtable,
#endif
#if HAVE_TLS
  &tport_tls_client_vtable,
  &tport_tls_vtable,
#endif
#if HAVE_SCTP		/* SCTP is broken */
  &tport_sctp_client_vtable,
  &tport_sctp_vtable,
#endif
  &tport_tcp_client_vtable,
  &tport_tcp_vtable,
  &tport_udp_client_vtable,
  &tport_udp_vtable,
#if 0
  &tport_threadpool_vtable,
#endif
#if HAVE_SOFIA_STUN
  &tport_stun_vtable,
#endif
};

/** Register new transport vtable */
int tport_register_type(tport_vtable_t const *vtp)
{
  int i;

  for (i = TPORT_NUMBER_OF_TYPES; i >= 0; i--) {
    if (tport_vtables[i] == NULL) {
      tport_vtables[i] = vtp;
      return 0;
    }
  }

  su_seterrno(ENOMEM);
  return -1;
}

/**Get a vtable for given protocol */
tport_vtable_t const *tport_vtable_by_name(char const *protoname,
					   enum tport_via public) 
{
  int i;

  for (i = TPORT_NUMBER_OF_TYPES; i >= 0; i--) {
    tport_vtable_t const *vtable = tport_vtables[i];

    if (vtable == NULL)
      continue;
    if (vtable->vtp_public != public)
      continue;
    if (strcasecmp(vtable->vtp_name, protoname))
      continue;

    assert(vtable->vtp_pri_size >= sizeof (tport_primary_t));
    assert(vtable->vtp_secondary_size >= sizeof (tport_t));
    
    return vtable;
  }

  return NULL;
}

#if 0
tport_set_f const *tport_set_methods[TPORT_NUMBER_OF_TYPES + 1] = 
  {
    tport_server_bind_set,
    tport_client_bind_set,
    tport_threadpool_set,
#if HAVE_SOFIA_NTH
    tport_http_connect_set,
#endif
#if HAVE_TLS
    tport_tls_set,
#endif
    NULL
  };

int tport_bind_set(tport_master_t *mr, 
		   tp_name_t const *tpn,
		   char const * const transports[],
		   tagi_t const *taglist,
		   tport_set_t **return_set,
		   int set_size)
{
  int i;

  for (i = TPORT_NUMBER_OF_TYPES; i >= 0; i--) {
    tport_set_f const *perhaps = tport_vtables[i];
    int result;

    if (perhaps == NULL)
      continue;

    result = perhaps(mr, tpn, transports, taglist, return_set, set_size);
    if (result != 0)
      return result;
  }

  return 0;
}
#endif

/** Bind transport objects.
 *
 * @param self        pointer to a transport object
 * @param tpn         desired transport address
 * @param transports  list of protocol names supported by stack
 * @param tag,value,... tagged argument list
 *
 * @TAGS
 * TPTAG_SERVER(), TPTAG_PUBLIC(), TPTAG_IDENT(), TPTAG_HTTP_CONNECT(),
 * TPTAG_CERTIFICATE(), TPTAG_TLS_VERSION(), and tags used with
 * tport_set_params(), especially TPTAG_QUEUESIZE().
 */
int tport_tbind(tport_t *self,
		tp_name_t const *tpn,
		char const * const transports[],
		tag_type_t tag, tag_value_t value, ...)
{
  ta_list ta;
  int server = 1, retval, public = 0;
  tp_name_t mytpn[1];
  tport_master_t *mr;
  char const *http_connect = NULL;
  
  if (self == NULL || tport_is_secondary(self) ||
      tpn == NULL || transports == NULL) {
    su_seterrno(EINVAL);
    return -1;
  }

  *mytpn = *tpn;

  if (mytpn->tpn_ident == NULL)
    mytpn->tpn_ident = self->tp_ident;

  ta_start(ta, tag, value);

  tl_gets(ta_args(ta),
	  TPTAG_SERVER_REF(server),
	  TPTAG_PUBLIC_REF(public),
	  TPTAG_IDENT_REF(mytpn->tpn_ident),
	  TPTAG_HTTP_CONNECT_REF(http_connect),
	  TAG_END());

  mr = self->tp_master; assert(mr);

  if (http_connect && public == 0)
    public = tport_type_connect;
    
  if (public && public != tport_type_stun)
    server = 0;

  if (server)
    retval = tport_bind_server(mr, mytpn, transports, public, ta_args(ta));
  else
    retval = tport_bind_client(mr, mytpn, transports, public, ta_args(ta));

  ta_end(ta);

  return retval;
}


/** Bind primary transport objects used by a client-only application.
 * @internal
 */
int tport_bind_client(tport_master_t *mr,
                      tp_name_t const *tpn,
                      char const * const transports[],
		      enum tport_via public,
		      tagi_t *tags)
{
  int i;
  tport_primary_t *pri = NULL, **tbf;
  tp_name_t tpn0[1] = {{ "*", "*", "*", "*", NULL, NULL }};
  char const *why = "unknown";

  tport_vtable_t const *vtable;

  if (public == tport_type_local)
    public = tport_type_client;
  
  SU_DEBUG_5(("%s(%p) to " TPN_FORMAT "\n",
	      __func__, (void *)mr, TPN_ARGS(tpn)));

  memset(tpn0, 0, sizeof(tpn0));

  for (tbf = &mr->mr_primaries; *tbf; tbf = &(*tbf)->pri_next)
    ;