nea_server.c

sip协议栈

				     NTATAG_METHOD("SUBSCRIBE"),
				     URLTAG_URL(url),
				     TAG_END());
    }
				     
    nes->nes_eventlist = eventlist; /* Every event is a list */
    if (eventlist && rq == NULL && rq_str == NULL)
      rq_str = "eventlist";

    if (rq)
      nes->nes_require = sip_require_dup(nes->nes_home, rq);
    else if (rq_str)
      nes->nes_require = sip_require_make(nes->nes_home, rq_str);

    nes->nes_timer = su_timer_create(su_root_task(nes->nes_root), 
				     nes->nes_min_throttle
				     ? 500L * nes->nes_min_throttle 
				     : 500L);

    if (nes->nes_allow_events &&
	nes->nes_eventity_uri &&
	(nes->nes_leg || leg == NULL) &&
	nes->nes_timer) {
      SU_DEBUG_5(("nea_server_create(%p): success\n", nes));
      su_timer_set(nes->nes_timer, nes_event_timer, nes);

      nes->nes_callback = callback;
      nes->nes_context = context;
    }
    else {
      SU_DEBUG_5(("nea_server_create(%p): failed\n", nes));
      nea_server_destroy(nes), nes = NULL;
    }
  }

  return nes;
}

/** Invoke the new event callback.
 *
 * The function nes_event_callback() calls the callback provided by the 
 * application using the notifier object. 
 *
 * @param nes pointer to notifier object
 * @param ev  pointer to event view
 * @param s   pointer to subscription object
 * @param sip pointer to subscribe request
 *
 * @return
 * The function nes_event_callback() returns -1 if the notifier object
 * has been destroyed by the callback function, 0 otherwise.
 */
static
int nes_new_event_callback(nea_server_t *nes, 
			   nea_event_t **ev_p, 
			   nea_event_view_t **view_p, 
			   nta_incoming_t *irq,
			   sip_t const *sip)
{
  if (nes->nes_callback)
    return nes->nes_callback(nes->nes_context, nes, ev_p, view_p, irq, sip);
  else
    return -1;
}

/** Shutdown event server.
 */
int nea_server_shutdown(nea_server_t *nes,
			int retry_after)
{
  nea_sub_t *s;
  int status = 200;
  int in_callback;

  if (nes == NULL)
    return 500;

  if (nes->nes_in_callback) {
    SU_DEBUG_5(("nea_server_shutdown(%p) while in callback\n", nes));
    return 100;
  }
  
  SU_DEBUG_5(("nea_server_shutdown(%p)\n", nes));

  in_callback = nes->nes_in_callback; nes->nes_in_callback = 1;

  for (s = nes->nes_subscribers; s; s = s->s_next) {
    if (s->s_state == nea_terminated)
      continue;
    if (s->s_pending_flush)
      continue;
    if (s->s_oreq == NULL)
      nea_sub_auth(s, nea_terminated, 
		   TAG_IF(retry_after, NEATAG_REASON("probation")),
		   TAG_IF(!retry_after, NEATAG_REASON("deactivated")),
		   TAG_IF(retry_after, NEATAG_RETRY_AFTER(retry_after)),
		   TAG_END());
    else
      status = 180;
  }    

  nes->nes_in_callback = in_callback;   

  return 200;
}

void nea_server_destroy(nea_server_t *nes)
{
  if (nes == NULL)
    return;

  if (nes->nes_in_callback) {
    SU_DEBUG_5(("nea_server_destroy(%p) while in callback\n", nes));
    nes->nes_pending_destroy = 1;
    return;
  }
  
  SU_DEBUG_5(("nea_server_destroy(%p)\n", nes));
  
  nta_leg_destroy(nes->nes_leg), nes->nes_leg = NULL;
  
  while (nes->nes_subscribers)
    nea_sub_destroy(nes->nes_subscribers);
  
  su_timer_destroy(nes->nes_timer), nes->nes_timer = NULL;
  
  su_home_unref(nes->nes_home);
}

/* ----------------------------------------------------------------- */

/**Update server payload.
 *
 * A nea event server has typed content that is delivered to the
 * subscribers. Different content types are each assigned a separate primary
 * view. There can be also primary views with "fake" content, content
 * delivered to politely blocked subscribers. 
 *
 * In addition to primary views, there can be secondary views, views
 * assigned to a single subscriber only. 
 *
 * @TAGS
 * The following tagged arguments are accepted:
 * <dl>
 *
 * <dt>SIPTAG_PAYLOAD() or SIPTAG_PAYLOAD_STR()
 * <dd>Updated event content.
 *
 * <dt>SIPTAG_CONTENT_TYPE() or SIPTAG_CONTENT_TYPE_STR().
 * <dd>MIME type of the content.
 *
 * <dt>NEATAG_FAKE(fak)
 * <dd>If @a fake is true, 'fake' view is updated.
 *
 * <dt>NEATAG_VIEW(view) 
 * <dd>If included in tagged arguments, @a view is * updated. Used when
 * updating secondary view.
 *
 * <dt>NEATAG_VERSION(version)
 * <dd>The application-provided @a version for
 * event content. After updated content has been sent to subscriber, @a
 * version is copied to subscriber information structure.
 *
 * <dt>NEATAG_EVMAGIC(context)
 * <dd>Application-provided @a context pointer.
 * The @a context pointer is returned by nea_view_magic() function.
 *
 * <dt>NEATAG_RELIABLE(reliable) 
 * <dd>The @a reliable flag determines how overlapping updates are handled. 
 * If @a reliable is true, all updates are delivered to the subscribers.
 *
 * <dt>NEATAG_THROTTLE(throttl)
 * <dd>Default value for event throttle for updated event view. Throttle
 * determines the minimum interval in seconds betweeen notifications. Note
 * that the notification indicating that the subscription has terminated
 * will be sent regardless of throttle.
 * 
 * The default throttle value is used if the subscriber does not include
 * a throttle parameter in @ref sip_event "Event" header of SUBSCRIBE request.
 * 
 * <dt>NEATAG_MINTHROTTLE()
 * <dd>Minimum allowed throttle value for updated event view.
 *
 * </dl>
 *
 * @retval -1 upon an error.
 * @retval 0  if event document was not updated.
 * @retval 1  if event document was updated.
 */
int nea_server_update(nea_server_t *nes,
		      nea_event_t *ev,
		      tag_type_t tag,
		      tag_value_t value,
		      ...)
{
  nea_event_view_t *evv = NULL;
  int fake = 0, updated;

  ta_list ta;

  if (ev == NULL)
    ev = nes->nes_events;

  ta_start(ta, tag, value);

  tl_gets(ta_args(ta),
	  NEATAG_FAKE_REF(fake),
	  NEATAG_VIEW_REF(evv),
	  TAG_NULL());
  
  updated = nea_view_update(nes, ev, &evv, 0, fake, ta_tags(ta));

  ta_end(ta);

  return updated;
}

static
int nea_view_update(nea_server_t *nes,
		    nea_event_t *ev,
		    nea_event_view_t **evvp,
		    int private,
		    int fake,
		    tag_type_t tag,
		    tag_value_t value,
		    ...)
{
  ta_list ta;

  su_home_t *home = nes->nes_home;

  sip_content_type_t const *ct = NULL;
  char const *cts = NULL, *pls = NULL;
  sip_payload_t const *pl = NULL;
  sip_payload_t *new_pl;
  nea_event_view_t *evv, **eevv = &evv;
  nea_event_view_t *primary = NULL, **primary_p = &primary;
  unsigned version = UINT_MAX;
  nea_evmagic_t *evmagic = NULL;
  int reliable = ev->ev_reliable;
  unsigned throttle = ev->ev_throttle;
  unsigned min_throttle = ev->ev_min_throttle;

  nea_event_queue_t evq[1] = {{ NULL }};

  ta_start(ta, tag, value);

  tl_gets(ta_args(ta),
	  SIPTAG_CONTENT_TYPE_REF(ct),
	  SIPTAG_CONTENT_TYPE_STR_REF(cts),
	  SIPTAG_PAYLOAD_REF(pl),
	  SIPTAG_PAYLOAD_STR_REF(pls),
	  NEATAG_VERSION_REF(version),
	  NEATAG_EVMAGIC_REF(evmagic),
	  NEATAG_RELIABLE_REF(reliable),
	  NEATAG_THROTTLE_REF(throttle),
	  NEATAG_MINTHROTTLE_REF(min_throttle),
	  TAG_NULL());
  
  ta_end(ta);

  if (min_throttle < throttle)
    min_throttle = throttle;

  if (ct == NULL && cts == NULL)
    return -1;

  if (ct)
    cts = ct->c_type;

  evv = *evvp;

  if (!evv) {
    int i;

    /* Check if the payload type already exists */
    for (i = 0; (evv = ev->ev_views[i]); i++)
      if (str0casecmp(cts, evv->evv_content_type->c_type) == 0)
	break;
    
    if (private && evv == NULL) /* No private view without primary view. */
      return -1;

    if (i == NEA_VIEW_MAX)	/* Too many primary views. */
      return -1;

    primary_p = eevv = ev->ev_views + i;
    
    /* Search for fakeness/eventlist/private view */
    if (evv && (private || evv->evv_private || evv->evv_fake != fake)) {
      for (eevv = &evv->evv_next; (evv = *eevv); eevv = &evv->evv_next) {
	if (private || evv->evv_private)
	  continue;
	if (evv->evv_fake == fake)
	  break;
      }
    }
  }
   
  /* New event view, allocate and link to chain */
  if (!evv) {
    sip_content_type_t *new_ct;

    evv = su_zalloc(home, sizeof (*evv));
    if (!evv)
      return -1;
    
    new_pl = pl ? sip_payload_dup(home, pl)
      : sip_payload_make(home, pls);
    
    new_ct = ct ? sip_content_type_dup(home, ct) 
      : sip_content_type_make(home, cts);
    
    if ((!new_pl && pl) || !new_ct) {
      su_free(home, evv); su_free(home, new_pl);
      return -1;
    }

    *evvp = *eevv = evv;
      
    evv->evv_primary = *primary_p;
    evv->evv_private = private != 0;
    evv->evv_fake = fake != 0;
    evv->evv_reliable = reliable != 0;
    evv->evv_magic = evmagic;
    evv->evv_content_type = new_ct;
    evv->evv_payload = new_pl;
    evv->evv_throttle = throttle;
    evv->evv_min_throttle = min_throttle;

    assert(evv->evv_content_type);
  }
  else {
    if (pl && 
	evv->evv_payload && 
	evv->evv_payload->pl_len == pl->pl_len &&
	memcmp(evv->evv_payload->pl_data, pl->pl_data, pl->pl_len) == 0)
      return 0;
    if (!pl && pls && evv->evv_payload && 
	evv->evv_payload->pl_len == strlen(pls) && 
	memcmp(evv->evv_payload->pl_data, pls, evv->evv_payload->pl_len) == 0)
      return 0;
    if (!pl && !pls && !evv->evv_payload)
      return 0;

    *evq = *evv->evv_head;

    new_pl = pl ? sip_payload_dup(home, pl) : sip_payload_make(home, pls);

    if (!new_pl && (pl || pls))
      return -1;
    
    evv->evv_payload = new_pl;
  }

  if (version != UINT_MAX)
    evv->evv_version = version;

  if (!fake)
    evv->evv_updated = ++ev->ev_updated;

  if (evq->evq_content_type)
    nea_view_queue(nes, evv, evq);

  SU_DEBUG_7(("nea_server_update(%p): %s (%s)\n", 
	      nes, ev->ev_event->o_type, evv->evv_content_type->c_type));

  return 1;
}

nea_event_view_t *nea_view_create(nea_server_t *nes,
				  nea_event_t *ev,
				  nea_evmagic_t *magic,
				  tag_type_t tag,
				  tag_value_t value,
				  ...)
{
  nea_event_view_t *evv = NULL;
  ta_list ta;

  if (ev == NULL)
    return NULL;

  ta_start(ta, tag, value);

  nea_view_update(nes, ev, &evv, 1, 0, ta_tags(ta));
			 
  ta_end(ta);

  return evv;
}

void nea_view_destroy(nea_server_t *nes, nea_event_view_t *evv)
{
  nea_event_view_t **evvp;
  nea_sub_t *s;

  if (nes == NULL || evv == NULL || !evv->evv_private)
    return;

  assert(evv->evv_primary && evv != evv->evv_primary);

  for (evvp = &evv->evv_primary->evv_next; *evvp; evvp = &(*evvp)->evv_next)
    if (*evvp == evv) {
      *evvp = evv->evv_next;
      break;
    }

  for (s = nes->nes_subscribers; s; s = s->s_next)
    if (s->s_view == evv)
      nea_sub_assign_view(s, evv->evv_primary);

  su_free(nes->nes_home, evv->evv_content_type);
  su_free(nes->nes_home, evv->evv_payload);
  su_free(nes->nes_home, evv);
}

nea_evmagic_t *nea_view_magic(nea_event_view_t const *evv)
{
  return evv ? evv->evv_magic : NULL;
}

void nea_view_set_magic(nea_event_view_t *evv, nea_evmagic_t *magic)
{
  if (evv)
    evv->evv_magic = magic;
}

unsigned nea_view_version(nea_event_view_t const *evv)
{
  return evv ? evv->evv_version : 0;
}

/** Get primary, non-fake event view for given content type  */
nea_event_view_t *nea_event_view(nea_event_t *ev, char const *content_type)
{
  int i;
  nea_event_view_t *evv;

  /* Check if the payload type already exists */
  for (i = 0; ev->ev_views[i]; i++)
    if (str0casecmp(content_type, ev->ev_views[i]->evv_content_type->c_type) == 0)
      break;

  for (evv = ev->ev_views[i]; evv; evv = evv->evv_next)
    if (!evv->evv_fake)
      return evv;

  return ev->ev_views[i];
}

/** Get the content type for event view */
sip_content_type_t const *nea_view_content_type(nea_event_view_t const *evv)
{
  return evv ? evv->evv_content_type : NULL;
}


/** Queue an old notification if needed. */
static
int nea_view_queue(nea_server_t *nes, 
		   nea_event_view_t *evv, 
		   nea_event_queue_t *evq)
{
  nea_sub_t *s = NULL;

  assert(nes && evv && evq);

  if (evv->evv_reliable)
    for (s = nes->nes_subscribers; s; s = s->s_next) {
      if (s->s_view != evv)
	continue;
      if (s->s_updated > evq->evq_updated)
	continue;
      if (s->s_updated == evq->evq_updated && s->s_oreq == NULL)