sipevent: added notifier

include/re_sipevent.h

@@ -5,6 +5,39 @@
  */
 
 
+/* Message Components */
+
+struct sipevent_event {
+	struct pl event;
+	struct pl params;
+	struct pl id;
+};
+
+enum sipevent_subst {
+	SIPEVENT_ACTIVE = 0,
+	SIPEVENT_PENDING,
+	SIPEVENT_TERMINATED,
+};
+
+enum sipevent_reason {
+	SIPEVENT_TIMEOUT = 0,
+	SIPEVENT_NORESOURCE,
+};
+
+struct sipevent_substate {
+	enum sipevent_subst state;
+	struct pl params;
+	struct pl expires;
+	struct pl reason;
+};
+
+int sipevent_event_decode(struct sipevent_event *se, const struct pl *pl);
+int sipevent_substate_decode(struct sipevent_substate *ss,
+			     const struct pl *pl);
+const char *sipevent_substate_name(enum sipevent_subst state);
+const char *sipevent_reason_name(enum sipevent_reason reason);
+
+
 /* Listener Socket */
 
 struct sipevent_sock;
@@ -14,6 +47,23 @@ int sipevent_listen(struct sipevent_sock **sockp, struct sip *sip,
 		    sip_msg_h *subh, void *arg);
 
 
+/* Notifier */
+
+struct sipnot;
+
+typedef void (sipevent_close_h)(int err, const struct sip_msg *msg, void *arg);
+
+int sipevent_accept(struct sipnot **notp, struct sipevent_sock *sock,
+		    const struct sip_msg *msg, struct sip_dialog *dlg,
+		    const struct sipevent_event *event,
+		    uint16_t scode, const char *reason, uint32_t expires_max,
+		    const char *cuser, const char *ctype,
+		    sip_auth_h *authh, void *aarg, bool aref,
+		    sipevent_close_h *closeh, void *arg, const char *fmt, ...);
+int sipevent_notify(struct sipnot *not, struct mbuf *mb);
+int sipevent_notifyf(struct sipnot *not, const char *fmt, ...);
+
+
 /* Subscriber */
 
 struct sipsub;
@@ -22,8 +72,6 @@ typedef int  (sipevent_fork_h)(struct sipsub **subp, struct sipsub *osub,
 			       const struct sip_msg *msg, void *arg);
 typedef void (sipevent_notify_h)(struct sip *sip, const struct sip_msg *msg,
 				 void *arg);
-typedef void (sipevent_close_h)(int err, const struct sip_msg *msg, void *arg);
-
 
 int sipevent_subscribe(struct sipsub **subp, struct sipevent_sock *sock,
 		       const char *uri, const char *from_name,
@@ -61,30 +109,3 @@ int sipevent_fork(struct sipsub **subp, struct sipsub *osub,
 		  sip_auth_h *authh, void *aarg, bool aref,
 		  sipevent_notify_h *notifyh, sipevent_close_h *closeh,
 		  void *arg);
-
-
-/* Message Components */
-
-struct sipevent_event {
-	struct pl event;
-	struct pl params;
-	struct pl id;
-};
-
-enum sipevent_subst {
-	SIPEVENT_ACTIVE = 0,
-	SIPEVENT_PENDING,
-	SIPEVENT_TERMINATED,
-};
-
-struct sipevent_substate {
-	enum sipevent_subst state;
-	struct pl params;
-	struct pl expires;
-	struct pl reason;
-};
-
-int sipevent_event_decode(struct sipevent_event *se, const struct pl *pl);
-int sipevent_substate_decode(struct sipevent_substate *ss,
-			     const struct pl *pl);
-const char *sipevent_substate_name(enum sipevent_subst state);

src/sipevent/listen.c

@@ -65,10 +65,11 @@ static bool event_cmp(const struct sipevent_event *evt,
 
 static bool not_cmp_handler(struct le *le, void *arg)
 {
-	const struct sip_msg *msg = arg;
+	const struct subcmp *cmp = arg;
 	struct sipnot *not = le->data;
 
-	return sip_dialog_cmp(not->dlg, msg);
+	return sip_dialog_cmp(not->dlg, cmp->msg) &&
+		event_cmp(cmp->evt, not->event, not->id, -1);
 }
 
 
@@ -96,11 +97,17 @@ static bool sub_cmp_half_handler(struct le *le, void *arg)
 
 
 static struct sipnot *sipnot_find(struct sipevent_sock *sock,
-				   const struct sip_msg *msg)
+				  const struct sip_msg *msg,
+				  const struct sipevent_event *evt)
 {
+	struct subcmp cmp;
+
+	cmp.msg = msg;
+	cmp.evt = evt;
+
 	return list_ledata(hash_lookup(sock->ht_not,
 				       hash_joaat_pl(&msg->callid),
-				       not_cmp_handler, (void *)msg));
+				       not_cmp_handler, &cmp));
 }
 
 
@@ -229,10 +236,19 @@ static void notify_handler(struct sipevent_sock *sock,
 static void subscribe_handler(struct sipevent_sock *sock,
 			      const struct sip_msg *msg)
 {
+	struct sipevent_event event;
 	struct sip *sip = sock->sip;
+	const struct sip_hdr *hdr;
 	struct sipnot *not;
+	uint32_t expires;
 
-	not = sipnot_find(sock, msg);
+	hdr = sip_msg_hdr(msg, SIP_HDR_EVENT);
+	if (!hdr || sipevent_event_decode(&event, &hdr->val)) {
+		(void)sip_reply(sip, msg, 400, "Bad Event Header");
+		return;
+	}
+
+	not = sipnot_find(sock, msg, &event);
 	if (!not || not->terminated) {
 		(void)sip_reply(sip, msg, 481, "Subscription Does Not Exist");
 		return;
@@ -245,7 +261,18 @@ static void subscribe_handler(struct sipevent_sock *sock,
 
 	(void)sip_dialog_update(not->dlg, msg);
 
-	/* todo: implement notifier */
+	if (pl_isset(&msg->expires))
+		expires = pl_u32(&msg->expires);
+	else
+		expires = DEFAULT_EXPIRES;
+
+	sipnot_refresh(not, expires);
+
+	(void)sipnot_reply(not, msg, 200, "OK");
+
+	if (expires > 0) {
+		(void)sipnot_notify(not);
+	}
 }
 
 

src/sipevent/mod.mk

@@ -6,4 +6,5 @@
 
 SRCS	+= sipevent/listen.c
 SRCS	+= sipevent/msg.c
+SRCS	+= sipevent/notify.c
 SRCS	+= sipevent/subscribe.c

src/sipevent/msg.c

@@ -81,3 +81,14 @@ const char *sipevent_substate_name(enum sipevent_subst state)
 	default:                  return "???";
 	}
 }
+
+
+const char *sipevent_reason_name(enum sipevent_reason reason)
+{
+	switch (reason) {
+
+	case SIPEVENT_TIMEOUT:    return "timeout";
+	case SIPEVENT_NORESOURCE: return "noresource";
+	default:                  return "???";
+	}
+}

src/sipevent/notify.c

@@ -0,0 +1,441 @@
+/**
+ * @file not.c  SIP Event Notify
+ *
+ * Copyright (C) 2010 Creytiv.com
+ */
+#include <string.h> // todo: remove
+#include <re_types.h>
+#include <re_mem.h>
+#include <re_mbuf.h>
+#include <re_sa.h>
+#include <re_list.h>
+#include <re_hash.h>
+#include <re_fmt.h>
+#include <re_uri.h>
+#include <re_sys.h>
+#include <re_tmr.h>
+#include <re_sip.h>
+#include <re_sipevent.h>
+#include "sipevent.h"
+
+
+static int notify_request(struct sipnot *not, bool reset_ls);
+
+
+static void internal_close_handler(int err, const struct sip_msg *msg,
+				   void *arg)
+{
+	(void)err;
+	(void)msg;
+	(void)arg;
+}
+
+
+static bool terminate(struct sipnot *not, enum sipevent_reason reason)
+{
+	not->terminated = true;
+	not->reason     = reason;
+	not->closeh     = internal_close_handler;
+
+	if (not->req) {
+		mem_ref(not);
+		return true;
+	}
+
+	if (not->subscribed && !notify_request(not, true)) {
+		mem_ref(not);
+		return true;
+	}
+
+	return false;
+}
+
+
+static void destructor(void *arg)
+{
+	struct sipnot *not = arg;
+
+	tmr_cancel(&not->tmr);
+
+	if (!not->terminated) {
+
+		if (terminate(not, SIPEVENT_NORESOURCE))
+			return;
+	}
+
+	hash_unlink(&not->he);
+	mem_deref(not->req);
+	mem_deref(not->dlg);
+	mem_deref(not->auth);
+	mem_deref(not->mb);
+	mem_deref(not->event);
+	mem_deref(not->id);
+	mem_deref(not->cuser);
+	mem_deref(not->hdrs);
+	mem_deref(not->ctype);
+	mem_deref(not->sock);
+	mem_deref(not->sip);
+}
+
+
+static void sipnot_terminate(struct sipnot *not, int err,
+			     const struct sip_msg *msg,
+			     enum sipevent_reason reason)
+{
+	sipevent_close_h *closeh;
+	void *arg;
+
+	closeh = not->closeh;
+	arg    = not->arg;
+
+	tmr_cancel(&not->tmr);
+	(void)terminate(not, reason);
+
+	closeh(err, msg, arg);
+}
+
+
+static void tmr_handler(void *arg)
+{
+	struct sipnot *not = arg;
+
+	re_printf("subscription expired\n");
+
+	sipnot_terminate(not, ETIMEDOUT, NULL, SIPEVENT_TIMEOUT);
+}
+
+
+void sipnot_refresh(struct sipnot *not, uint32_t expires)
+{
+	expires = min(expires, not->expires_max);
+
+	re_printf("will expire in %u secs\n", expires);
+
+	tmr_start(&not->tmr, expires * 1000, tmr_handler, not);
+}
+
+
+static void response_handler(int err, const struct sip_msg *msg, void *arg)
+{
+	struct sipnot *not = arg;
+
+	if (err)
+		re_printf("notify reply: %s\n", strerror(err));
+	else
+		re_printf("notify reply: %u %r\n", msg->scode, &msg->reason);
+
+	if (err || sip_request_loops(&not->ls, msg->scode))
+		goto out;
+
+	if (msg->scode < 200) {
+		return;
+	}
+	else if (msg->scode < 300) {
+
+		(void)sip_dialog_update(not->dlg, msg);
+	}
+	else {
+		switch (msg->scode) {
+
+		case 401:
+		case 407:
+			err = sip_auth_authenticate(not->auth, msg);
+			if (err) {
+				err = (err == EAUTH) ? 0 : err;
+				break;
+			}
+
+			err = notify_request(not, false);
+			if (err)
+				break;
+
+			return;
+
+		case 403:
+			sip_auth_reset(not->auth);
+			break;
+
+		case 481:
+			not->subscribed = false;
+			break;
+		}
+	}
+
+ out:
+	if (not->termsent) {
+		mem_deref(not);
+	}
+	else if (not->terminated) {
+		if (!not->subscribed || notify_request(not, true))
+			mem_deref(not);
+	}
+	else if (!not->subscribed) {
+		sipnot_terminate(not, err, msg, -1);
+	}
+	else if (not->notify_pending) {
+		(void)notify_request(not, true);
+	}
+}
+
+
+static int send_handler(enum sip_transp tp, const struct sa *src,
+			const struct sa *dst, struct mbuf *mb, void *arg)
+{
+	struct sipnot *not = arg;
+	(void)dst;
+
+	return mbuf_printf(mb, "Contact: <sip:%s@%J%s>\r\n",
+                           not->cuser, src, sip_transp_param(tp));
+}
+
+
+static int print_event(struct re_printf *pf, const struct sipnot *not)
+{
+	if (not->id)
+		return re_hprintf(pf, "%s;id=%s", not->event, not->id);
+	else
+		return re_hprintf(pf, "%s", not->event);
+}
+
+
+static int print_substate(struct re_printf *pf, const struct sipnot *not)
+{
+	if (not->terminated) {
+		return re_hprintf(pf, "terminated;reason=%s",
+				  sipevent_reason_name(not->reason));
+	}
+	else {
+		uint32_t expires;
+
+		expires = (uint32_t)(tmr_get_expire(&not->tmr) / 1000);
+
+		return re_hprintf(pf, "active;expires=%u", expires);
+	}
+}
+
+
+static int print_content(struct re_printf *pf, const struct sipnot *not)
+{
+	if (!not->mb)
+		return re_hprintf(pf,
+				  "Content-Length: 0\r\n"
+				  "\r\n");
+	else
+		return re_hprintf(pf,
+				  "Content-Type: %s\r\n"
+				  "Content-Length: %zu\r\n"
+				  "\r\n"
+				  "%b",
+				  not->ctype,
+				  mbuf_get_left(not->mb),
+				  mbuf_buf(not->mb),
+				  mbuf_get_left(not->mb));
+}
+
+
+static int notify_request(struct sipnot *not, bool reset_ls)
+{
+	if (reset_ls)
+		sip_loopstate_reset(&not->ls);
+
+	if (not->terminated)
+		not->termsent = true;
+
+	not->notify_pending = false;
+
+	return sip_drequestf(&not->req, not->sip, true, "NOTIFY",
+			     not->dlg, 0, not->auth,
+			     send_handler, response_handler, not,
+			     "Event: %H\r\n"
+			     "Subscription-State: %H\r\n"
+			     "%s"
+			     "%H",
+			     print_event, not,
+			     print_substate, not,
+			     not->hdrs,
+			     print_content, not);
+}
+
+
+int sipnot_notify(struct sipnot *not)
+{
+	if (not->req) {
+		not->notify_pending = true;
+		return 0;
+	}
+
+	return notify_request(not, true);
+}
+
+
+int sipnot_reply(struct sipnot *not, const struct sip_msg *msg,
+		 uint16_t scode, const char *reason)
+{
+	uint32_t expires;
+
+	expires = (uint32_t)(tmr_get_expire(&not->tmr) / 1000);
+
+	return sip_treplyf(NULL, NULL, not->sip, msg, true, scode, reason,
+			   "Contact: <sip:%s@%J%s>\r\n"
+			   "Expires: %u\r\n"
+			   "Content-Length: 0\r\n"
+			   "\r\n",
+			   not->cuser, &msg->dst, sip_transp_param(msg->tp),
+			   expires);
+}
+
+
+int sipevent_accept(struct sipnot **notp, struct sipevent_sock *sock,
+		    const struct sip_msg *msg, struct sip_dialog *dlg,
+		    const struct sipevent_event *event,
+		    uint16_t scode, const char *reason, uint32_t expires_max,
+		    const char *cuser, const char *ctype,
+		    sip_auth_h *authh, void *aarg, bool aref,
+		    sipevent_close_h *closeh, void *arg, const char *fmt, ...)
+{
+	struct sipnot *not;
+	uint32_t expires;
+	int err;
+
+	if (!notp || !sock || !msg || !scode || !reason || !expires_max ||
+	    !cuser || !ctype)
+		return EINVAL;
+
+	not = mem_zalloc(sizeof(*not), destructor);
+	if (!not)
+		return ENOMEM;
+
+	if (!pl_strcmp(&msg->met, "REFER")) {
+
+		err = str_dup(&not->event, "refer");
+		if (err)
+			goto out;
+
+		err = re_sdprintf(&not->id, "%u", msg->cseq.num);
+		if (err)
+			goto out;
+	}
+	else {
+		if (!event) {
+			err = EINVAL;
+			goto out;
+		}
+
+		err = pl_strdup(&not->event, &event->event);
+		if (err)
+			goto out;
+
+		if (pl_isset(&event->id)) {
+
+			err = pl_strdup(&not->id, &event->id);
+			if (err)
+				goto out;
+		}
+	}
+
+	if (dlg) {
+		not->dlg = mem_ref(dlg);
+	}
+	else {
+		err = sip_dialog_accept(&not->dlg, msg);
+		if (err)
+			goto out;
+	}
+
+	hash_append(sock->ht_not,
+		    hash_joaat_str(sip_dialog_callid(not->dlg)),
+		    &not->he, not);
+
+	err = sip_auth_alloc(&not->auth, authh, aarg, aref);
+	if (err)
+		goto out;
+
+	err = str_dup(&not->cuser, cuser);
+	if (err)
+		goto out;
+
+	err = str_dup(&not->ctype, ctype);
+	if (err)
+		goto out;
+
+	if (fmt) {
+		va_list ap;
+
+		va_start(ap, fmt);
+		err = re_vsdprintf(&not->hdrs, fmt, ap);
+		va_end(ap);
+		if (err)
+			goto out;
+	}
+
+	not->expires_max = expires_max;
+	not->sock   = mem_ref(sock);
+	not->sip    = mem_ref(sock->sip);
+	not->closeh = closeh ? closeh : internal_close_handler;
+	not->arg    = arg;
+
+	if (pl_isset(&msg->expires))
+		expires = pl_u32(&msg->expires);
+	else
+		expires = DEFAULT_EXPIRES;
+
+	sipnot_refresh(not, expires);
+
+	err = sipnot_reply(not, msg, scode, reason);
+	if (err)
+		goto out;
+
+	not->subscribed = true;
+
+ out:
+	if (err)
+		mem_deref(not);
+	else
+		*notp = not;
+
+	return err;
+}
+
+
+int sipevent_notify(struct sipnot *not, struct mbuf *mb)
+{
+	if (!not || not->terminated)
+		return EINVAL;
+
+	mem_deref(not->mb);
+	not->mb = mem_ref(mb);
+
+	return sipnot_notify(not);
+}
+
+
+int sipevent_notifyf(struct sipnot *not, const char *fmt, ...)
+{
+	struct mbuf *mb;
+	va_list ap;
+	int err;
+
+	if (!not || not->terminated || !fmt)
+		return EINVAL;
+
+	mb = mbuf_alloc(1024);
+	if (!mb)
+		return ENOMEM;
+
+	va_start(ap, fmt);
+	err = mbuf_vprintf(mb, fmt, ap);
+	va_end(ap);
+	if (err)
+		goto out;
+
+	mb->pos = 0;
+
+	err = sipevent_notify(not, mb);
+	if (err)
+		goto out;
+
+ out:
+	mem_deref(mb);
+
+	return err;
+}

src/sipevent/sipevent.h

@@ -4,6 +4,10 @@
  * Copyright (C) 2010 Creytiv.com
  */
 
+enum {
+	DEFAULT_EXPIRES = 3600,
+};
+
 
 /* Listener Socket */
 
@@ -21,10 +25,34 @@ struct sipevent_sock {
 
 struct sipnot {
 	struct le he;
+	struct sip_loopstate ls;
+	struct tmr tmr;
+	struct sipevent_sock *sock;
+	struct sip_request *req;
 	struct sip_dialog *dlg;
+	struct sip_auth *auth;
+	struct sip *sip;
+	struct mbuf *mb;
+	char *event;
+	char *id;
+	char *cuser;
+	char *hdrs;
+	char *ctype;
+	sipevent_close_h *closeh;
+	void *arg;
+	uint32_t expires_max;
+	enum sipevent_reason reason;
+	bool notify_pending;
+	bool subscribed;
 	bool terminated;
+	bool termsent;
 };
 
+void sipnot_refresh(struct sipnot *not, uint32_t expires);
+int  sipnot_notify(struct sipnot *not);
+int  sipnot_reply(struct sipnot *not, const struct sip_msg *msg,
+		  uint16_t scode, const char *reason);
+
 
 /* Subscriber */
 

src/sipevent/subscribe.c

@@ -20,7 +20,6 @@
 
 
 enum {
-	DEFAULT_EXPIRES = 3600,
 	RESUB_FAIL_WAIT = 60000,
 	RESUB_FAILC_MAX = 7,
 };
@@ -348,7 +347,7 @@ static int sipsub_alloc(struct sipsub **subp, struct sipevent_sock *sock,
 	struct sipsub *sub;
 	int err;
 
-	if (!subp || !sock || !event || !expires ||!cuser)
+	if (!subp || !sock || !event || !expires || !cuser)
 		return EINVAL;
 
 	if (!dlg && (!uri || !from_uri))