hop-2012/server/subscription.ml

(* Copyright 2012 Tony Garnock-Jones <tonygarnockjones@gmail.com>. *)

(* This file is part of Hop. *)

(* Hop is free software: you can redistribute it and/or modify it *)
(* under the terms of the GNU General Public License as published by the *)
(* Free Software Foundation, either version 3 of the License, or (at your *)
(* option) any later version. *)

(* Hop is distributed in the hope that it will be useful, but *)
(* WITHOUT ANY WARRANTY; without even the implied warranty of *)
(* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU *)
(* General Public License for more details. *)

(* You should have received a copy of the GNU General Public License *)
(* along with Hop.  If not, see <http://www.gnu.org/licenses/>. *)

open Lwt
open Datastructures

type t = {
    mutable live: bool;
    uuid: Uuid.t;
    filter: Sexp.t;
    sink: Node.name;
    name: Sexp.t
  }

type creation_t =
  | Old of t
  | New of t

type set_t = {
  mutable subscription_table: Uuid.t SexpMap.t;
  mutable uuid_table: t BytesMap.t
}

let new_set () = {
  subscription_table = SexpMap.empty;
  uuid_table = BytesMap.empty
}

let count subs = SexpMap.cardinal subs.subscription_table

let key_from sink_bs name filter = Sexp.Arr [Sexp.Str sink_bs; name; filter]

let create source subs filter sink_str name reply_sink reply_name =
  let key = key_from sink_str name filter in
  try
    let uuid = SexpMap.find key subs.subscription_table in
    lwt () =
      Node.post_ignore' reply_sink reply_name (Message.subscribe_ok (Sexp.str uuid)) Sexp.emptystr
    in
    return (Old (BytesMap.find (Bytes.of_string uuid) subs.uuid_table))
  with Not_found ->
    let uuid = Uuid.create () in
    let sink = Node.name_of_bytes sink_str in
    let sub = {
      live = true;
      uuid = uuid;
      filter = filter;
      sink = sink;
      name = name
    } in
    subs.uuid_table <- BytesMap.add (Bytes.of_string uuid) sub subs.uuid_table;
    subs.subscription_table <- SexpMap.add key uuid subs.subscription_table;
    lwt () = Lwt.join [
      Meta.announce_subscription source filter sink_str name true;
      Node.post_ignore' reply_sink reply_name (Message.subscribe_ok (Sexp.str uuid)) Sexp.emptystr
    ] in
    return (New sub)

let delete source subs uuid =
  try
    let sub = BytesMap.find uuid subs.uuid_table in
    sub.live <- false;
    subs.uuid_table <- BytesMap.remove uuid subs.uuid_table;
    let key = key_from sub.sink.Node.label sub.name sub.filter in
    subs.subscription_table <- SexpMap.remove key subs.subscription_table;
    lwt () = Meta.announce_subscription source sub.filter sub.sink.Node.label sub.name false in
    return (Some sub)
  with Not_found ->
    return None

let lookup subs uuid =
  try Some (BytesMap.find uuid subs.uuid_table)
  with Not_found -> None

let send_to_subscription' sub body delete_action =
  if not sub.live
  then return false
  else
    match_lwt Node.post sub.sink sub.name body (Sexp.str sub.uuid) with
      | true -> return true
      | false -> (lwt _ = delete_action sub.uuid in return false)

let send_to_subscription source subs sub body =
  send_to_subscription' sub body (fun (uuid) -> delete source subs (Bytes.of_string uuid))