tport.c

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

  if (addrlen < src->ai_addrlen)
    return -1;

  memcpy(dst, src, sizeof *dst);

  if (src->ai_addrlen < addrlen)
    memset(addr, 0, addrlen);

  dst->ai_addr = memcpy(addr, src->ai_addr, src->ai_addrlen);
  dst->ai_next = NULL;

  return 0;
}

/** Close a transport. 
 * 
 * Close the socket associated with a transport object. Report an error to
 * all pending clients, if required. Set/reset timer, too.
 */
void tport_close(tport_t *self)
{
  SU_DEBUG_5(("%s(%p): " TPN_FORMAT "\n",
	      __func__, (void *)self, TPN_ARGS(self->tp_name)));

  if (self->tp_closed || !tport_is_secondary(self))
    return;

  tprb_remove(&self->tp_pri->pri_open, self);
  tplist_insert(&self->tp_pri->pri_closed, self);

  self->tp_closed = 1;
  self->tp_send_close = 3;
  self->tp_recv_close = 3;

  if (self->tp_params->tpp_sdwn_error && self->tp_pused)
    tport_error_report(self, -1, NULL);

  if (self->tp_pri->pri_vtable->vtp_shutdown)
    self->tp_pri->pri_vtable->vtp_shutdown(self, 2);
  else if (self->tp_socket != -1)
    shutdown(self->tp_socket, 2);

  if (self->tp_index)
    su_root_deregister(self->tp_master->mr_root, self->tp_index);
  self->tp_index = 0;
#if SU_HAVE_BSDSOCK
  if (self->tp_socket != -1)
    su_close(self->tp_socket);
  self->tp_socket = -1;
#endif

  /* Zap the queued messages */
  if (self->tp_queue) {
    unsigned short i, N = self->tp_params->tpp_qsize;
    for (i = 0; i < N; i++) {
      if (self->tp_queue[i])
	msg_ref_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
    }
  }
  
  self->tp_index = 0;
  self->tp_events = 0;
}

/** Shutdown a transport.
 *
 * The tport_shutdown() shuts down a full-duplex transport connection
 * partially or completely. If @a how is 0, the further incoming data is
 * shut down. If @a how is 1, further outgoing data is shut down. If @a how
 * is 2, both incoming and outgoing traffic is shut down.
 *
 */
int tport_shutdown(tport_t *self, int how)
{
  int retval;
  if (!tport_is_secondary(self))
    return -1;
  retval = tport_shutdown0(self, how);
  tport_set_secondary_timer(self);
  return retval;
}

/** Internal shutdown function */
int tport_shutdown0(tport_t *self, int how)
{
  SU_DEBUG_7(("%s(%p, %d)\n", __func__, (void *)self, how));

  if (!tport_is_tcp(self) ||
      how < 0 || how >= 2 ||
      (how == 0 && self->tp_send_close) ||
      (how == 1 && self->tp_recv_close > 1)) {
    tport_close(self);
    return 1;
  }

  if (self->tp_pri->pri_vtable->vtp_shutdown)
    self->tp_pri->pri_vtable->vtp_shutdown(self, how);
  else
    shutdown(self->tp_socket, how);

  if (how == 0) {
    self->tp_recv_close = 2;
    tport_set_events(self, 0, SU_WAIT_IN);
    if (self->tp_params->tpp_sdwn_error && self->tp_pused)
      tport_error_report(self, -1, NULL);
  }
  else if (how == 1) {
    self->tp_send_close = 2;
    tport_set_events(self, 0, SU_WAIT_OUT);
    if (tport_has_queued(self)) {
      unsigned short i, N = self->tp_params->tpp_qsize;
      for (i = 0; i < N; i++) {
	if (self->tp_queue[i]) {
	  tport_pending_errmsg(self, self->tp_queue[i], EPIPE);
	  msg_ref_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
	}
      }
    }
  }

  return 0;
}

static void tport_secondary_timer(su_root_magic_t *magic,
				  su_timer_t *t,
				  tport_t *self)
{
  su_time_t now;

  if (tport_is_closed(self)) {
    if (self->tp_refs == 0)
      tport_zap_secondary(self);
    return;
  }

  now = /* su_timer_expired(t); */ su_now();

  if (self->tp_pri->pri_vtable->vtp_secondary_timer)
    self->tp_pri->pri_vtable->vtp_secondary_timer(self, now);
  else
    tport_base_timer(self, now);
}

/** Base timer for secondary transports.  
 *
 * Closes and zaps unused transports.  Sets the timer again.
 */
void tport_base_timer(tport_t *self, su_time_t now)
{
  unsigned timeout = self->tp_params->tpp_idle;

  if (timeout != UINT_MAX) {
    if (self->tp_refs == 0 && 
	self->tp_msg == NULL && 
	!tport_has_queued(self) &&
	su_time_cmp(su_time_add(self->tp_rtime, timeout), now) < 0 &&
	su_time_cmp(su_time_add(self->tp_stime, timeout), now) < 0) {
      SU_DEBUG_7(("%s(%p): unused for %d ms,%s zapping\n",
		  __func__, (void *)self,
		  timeout, tport_is_closed(self) ? "" : " closing and"));
      if (!tport_is_closed(self))
	tport_close(self);
      tport_zap_secondary(self);
      return;
    }
  }

  tport_set_secondary_timer(self);
}

/** Set timer for a secondary transport. 
 *
 * This function should be called after any network activity:
 * tport_base_connect(), tport_send_msg(), tport_send_queue(),
 * tport_recv_data(), tport_shutdown0(), tport_close(),
 *
 * @retval 0 always
 */
int tport_set_secondary_timer(tport_t *self)
{
  su_time_t const infinity = { ULONG_MAX, 999999 };
  su_time_t target = infinity;
  char const *why = "not specified";
  su_timer_f timer = tport_secondary_timer;

  if (!tport_is_secondary(self))
    return 0;

  if (tport_is_closed(self)) {
    if (self->tp_refs == 0) {
      SU_DEBUG_7(("tport(%p): set timer at %u ms because %s\n",
		  self, 0, "zap"));
      su_timer_set_interval(self->tp_timer, timer, self, 0);
    }
    else
      su_timer_reset(self->tp_timer);

    return 0;
  }

  if (self->tp_params->tpp_idle != UINT_MAX) {
    if (self->tp_refs == 0 && 
	self->tp_msg == NULL && !tport_has_queued(self)) {
      if (su_time_cmp(self->tp_stime, self->tp_rtime) < 0) {
	target = su_time_add(self->tp_rtime, self->tp_params->tpp_idle);
	why = "idle since recv";
      }
      else {
	target = su_time_add(self->tp_stime, self->tp_params->tpp_idle);
	why = "idle since send";
      }
    }
  }

  if (self->tp_pri->pri_vtable->vtp_next_secondary_timer)
    self->tp_pri->pri_vtable->
      vtp_next_secondary_timer(self, &target, &why);

  if (su_time_cmp(target, infinity)) {
    SU_DEBUG_7(("tport(%p): set timer at %ld ms because %s\n",
		(void *)self, su_duration(target, su_now()), why));
    su_timer_set_at(self->tp_timer, timer, self, target);
  }
  else {
    SU_DEBUG_9(("tport(%p): reset timer\n", (void *)self));
    su_timer_reset(self->tp_timer);
  }

  return 0;
}


/** Flush idle connections. */
int tport_flush(tport_t *tp)
{
  tport_t *tp_next;
  tport_primary_t *pri;

  if (tp == NULL)
    return -1;

  pri = tp->tp_pri;

  while (pri->pri_closed)
    tport_zap_secondary(pri->pri_closed);

  /* Go through all secondary transports, zap idle ones */
  for (tp = tprb_first(tp->tp_pri->pri_open); tp; tp = tp_next) {
    tp_next = tprb_succ(tp);

    if (tp->tp_refs != 0)
      continue;

    SU_DEBUG_1(("tport_flush(%p): %szapping\n",
		(void *)tp, tport_is_closed(tp) ? "" : "closing and "));

    tport_close(tp);
    tport_zap_secondary(tp);
  }

  return 0;
}

/**Convert sockaddr_t to a transport name.
 *
 * @retval 0 when successful
 * @retval -1 upon an error
 */
int tport_convert_addr(su_home_t *home,
		       tp_name_t *tpn,
		       char const *protoname,
		       char const *canon,
		       su_sockaddr_t const *su)
{
  tp_name_t name[1] = {{ NULL }};
  char const *host;
  char buf[TPORT_HOSTPORTSIZE];
  char port[8];
  size_t canonlen = canon ? strlen(canon) : 0;

  if (su == NULL)
    host = "*";
  else if (!SU_SOCKADDR_INADDR_ANY(su))
    host = tport_hostport(buf, sizeof(buf), su, 0);
  else if (canonlen && su->su_family == AF_INET && 
	   strspn(canon, "0123456789.") == canonlen)
    host = canon;
#if SU_HAVE_IN6
  else if (canonlen && su->su_family == AF_INET6 && 
	   strspn(canon, "0123456789abcdefABCDEF:.") == canonlen)
    host = canon;
#endif
  else
    host = localipname(su->su_family, buf, sizeof(buf));

  if (host == NULL)
    return -1;

  if (su == NULL)
    strcpy(port, "*");
  else
    snprintf(port, sizeof(port), "%u", ntohs(su->su_port));
  
  name->tpn_proto = protoname;
  name->tpn_host = host;
  name->tpn_canon = canon ? canon : host;
  name->tpn_port = port;

  return tport_name_dup(home, tpn, name);
}

/** Set transport object name. @internal
 */
static
int tport_setname(tport_t *self,
		  char const *protoname,
		  su_addrinfo_t const *ai,
		  char const *canon)
{
  su_addrinfo_t *selfai = self->tp_addrinfo;

  if (tport_convert_addr(self->tp_home, self->tp_name, 
			 protoname, canon, 
			 (su_sockaddr_t *)ai->ai_addr) < 0)
    return -1;

  if (tport_is_secondary(self))
    self->tp_ident = self->tp_pri->pri_primary->tp_ident;

  selfai->ai_flags = ai->ai_flags & TP_AI_MASK;

  selfai->ai_family = ai->ai_family;     
  selfai->ai_socktype = ai->ai_socktype; 
  selfai->ai_protocol = ai->ai_protocol; 
  selfai->ai_canonname = (char *)self->tp_name->tpn_canon;

  if (ai->ai_addr) {
    assert(ai->ai_family), assert(ai->ai_socktype), assert(ai->ai_protocol);
    memcpy(self->tp_addr, ai->ai_addr, selfai->ai_addrlen = ai->ai_addrlen);
  }

  return 0;
}

/**Resolve protocol name.
 *
 * Convert a protocol name to IP protocol number and socket type used by
 * su_getaddrinfo().
 *
 * @param hints hints with the protocol number and socktype [OUT]
 * @param proto protocol name [IN]
 * @param flags hint flags      
 */
static
int getprotohints(su_addrinfo_t *hints, 
		  char const *proto,
		  int flags)
{
  memset(hints, 0, sizeof *hints);

  hints->ai_flags = flags;
  hints->ai_canonname = (char *)proto;

#if HAVE_TLS
  if (strcasecmp(proto, "tls") == 0)
    proto = "tcp";
#endif

#if HAVE_SCTP  
  if (strcasecmp(proto, "sctp") == 0) {
    hints->ai_protocol = IPPROTO_SCTP;
    hints->ai_socktype = SOCK_STREAM;
    return 0;
  }
#endif

  if (strcasecmp(proto, "udp") == 0) {
    hints->ai_protocol = IPPROTO_UDP;
    hints->ai_socktype = SOCK_DGRAM;
    return 0;
  }
  
  if (strcasecmp(proto, "tcp") == 0) {
    hints->ai_protocol = IPPROTO_TCP;
    hints->ai_socktype = SOCK_STREAM;
    return 0;
  }

  return -1;
}

/** Get local IP.
 *
 * Get primary local IP address in URI format (IPv6 address will be
 * []-quoted).
 */
static
char *localipname(int pf, char *buf, size_t bufsiz)
{
  su_localinfo_t *li = NULL, hints[1] = {{ LI_NUMERIC | LI_CANONNAME }};
  size_t n;
  int error;

  hints->li_family = pf;

#if SU_HAVE_IN6
  if (pf == AF_INET6) {
    /* Link-local addresses are not usable on IPv6 */
    hints->li_scope = LI_SCOPE_GLOBAL | LI_SCOPE_SITE /* | LI_SCOPE_HOST */;
  }
#endif

  if ((error = su_getlocalinfo(hints, &li))) {
#if SU_HAVE_IN6
    if (error == ELI_NOADDRESS && pf == AF_INET6) {
      hints->li_family = AF_INET;
      error = su_getlocalinfo(hints, &li);
      if (error == ELI_NOADDRESS) {
	hints->li_family = AF_INET6; hints->li_scope |= LI_SCOPE_HOST;
	error = su_getlocalinfo(hints, &li);
      }
      if (error == ELI_NOADDRESS) {
	hints->li_family = AF_INET;
	error = su_getlocalinfo(hints, &li);
      }
    }
#endif
    if (error) {
      SU_DEBUG_1(("tport: su_getlocalinfo: %s\n", su_gli_strerror(error)));
      return NULL;
    }
  }

  assert(li); assert(li->li_canonname);

  n = strlen(li->li_canonname);

  if (li->li_family == AF_INET) {
    if (n >= bufsiz)
      return NULL;
    
    memcpy(buf, li->li_canonname, n + 1);
  }
  else {
    if (n + 2 >= bufsiz)
      return NULL;

    memcpy(buf + 1, li->li_canonname, n);
    buf[0] = '['; buf[++n] = ']'; buf[++n] = '\0';
  }
      
  su_freelocalinfo(li);
  
  return buf;
}

/** Process errors from transport. */
void tport_error_report(tport_t *self, int errcode, 
			su_sockaddr_t const *addr)
{
  char const *errmsg;

  if (errcode == 0)
    return;
  else if (errcode > 0)
    errmsg = su_strerror(errcode);
  else
    /* Should be something  like ENOTCONN */
    errcode = 0, errmsg = "stream closed";

  if (addr && addr->su_family == AF_UNSPEC)
    addr = NULL;

  /* Mark this connection as unusable */
  if (errcode > 0 && tport_has_connection(self))
    self->tp_reusable = 0;

  /* Report error */
  if (addr && tport_pending_error(self, addr, errcode))
    ;
  else if (tport_is_secondary(self) &&
	   tport_pending_error(self, NULL, errcode) > 0)
    ;
  else if (self->tp_master->mr_tpac->tpac_error) {
    char *dstname = NULL;
    char hp[TPORT_HOSTPORTSIZE];

    if (addr)
      dstname = tport_hostport(hp, sizeof hp, addr, 1);

    STACK_ERROR(self, errcode, dstname);
  }
  else {
    if (tport_is_primary(self))
      SU_DEBUG_3(("%s(%p): %s (with %s)\n", __func__, (void *)self, 
		  errmsg, self->tp_protoname));
    else
      SU_DEBUG_3(("%s(%p): %s (with %s/%s:%s)\n", __func__, (void *)self, 
		  errmsg, self->tp_protoname, self->tp_host, self->tp_port));
  }

  /* Close connection */
  if (!self->tp_closed && errcode > 0 && tport_has_connection(self))
    tport_close(self);
}

/** Accept a new connection.
 * 
 * The function tport_accept() accepts a new c