anchor.diff:        ok = remove_topic_from_pid(TopicUri,Pid,State),
         true = ets:delete(Ets,SubscriptionId),
         true = ets:delete(Ets,TopicUri),
  -      publish_metaevent({on_delete,TopicUri,SubscriptionId},State),
  +      publish_metaevent(on_unsubscribe,TopicUri,SessionId,SubscriptionId,State),
  +      publish_metaevent(on_delete,TopicUri,SessionId,SubscriptionId,State),
         ok;
       {true,NewSubs} ->
         ok = remove_topic_from_pid(TopicUri,Pid,State),
         true = ets:insert(Ets,T#topic{subscribers=NewSubs}),
  +      publish_metaevent(on_unsubscribe,TopicUri,SessionId,SubscriptionId,State),
         ok
     end.
   
  
positive.code: -spec subscribe(Pid:: pid(),TopicUri :: binary(), Options :: map(), Session::term(), State :: record(state)) ->
    {ok, ID::non_neg_integer()} | {error, Reason :: term()}.
  subscribe(Pid,TopicUri,Options,Session,#state{ets=Ets}=State) ->
    case maps:get(match,Options,exact) of
      prefix ->
        {error,not_supported};
      wildcard ->
        {error,not_supported};
      exact ->
        SubscriptionId = case ets:lookup(Ets,TopicUri) of
                           [#topic{id=SID,subscribers=Subs}=T] ->
                             NewSubs = [Pid|lists:delete(Pid,Subs)],
                             ets:insert(Ets,T#topic{subscribers=NewSubs}),
                             SID;
                           [] ->
                             {ok,SID} = create_topic(TopicUri,[Pid],State),
                             SID
                         end,
        ok = add_topic_to_pid(TopicUri,Pid,Session,State),
        {ok,SubscriptionId}
    end.
  
  
negative.code:       ok
    end.
  
  create_topic(Uri,Match,Pids,#state{ets=Ets}=State) ->
    ID = gen_id(),
    Created = erlang:universaltime(),
    Topic = #topic{uri=Uri,id=ID,match=Match,created=Created,subscribers=Pids},
    case ets:insert_new(Ets,[#id_topic{id=ID,topic=Uri},Topic]) of
      true ->
        {ok,ID,#{id => ID,
                 created => cowboy_clock:rfc1123(Created),
                 uri => Uri,
                 match => Match}};
      false ->
        create_topic(Uri,Match,Pids,State)
    end.
  
  add_topic_to_pid(Topic,Pid,Session,#state{ets=Ets}) ->
    SessionId = erwa_session:get_id(Session),
    case ets:lookup(Ets,Pid) of
      [#pid_info{topics=Topics} = PT] ->
        true = ets:insert(Ets,PT#pid_info{topics=[Topic|lists:delete(Topic,Topics)]}),
        ok;
      [] ->
        PT = #pid_info{pid=Pid,id=SessionId,topics=[Topic]},
        true = ets:insert_new(Ets,PT),
        ok
    end.
  
  
  remove_topic_from_pid(Topic,Pid,#state{ets=Ets}) ->
    [#pid_info{topics=Topics} = PT] = ets:lookup(Ets,Pid),
    case lists:delete(Topic,Topics) of
      [] ->
        true = ets:delete(Ets,Pid);
      NewTopics ->
        true = ets:insert(Ets,PT#pid_info{topics=NewTopics})
    end,
    ok.
  
  get_id_for_pid(Pid,#state{ets=Ets}) ->
    [#pid_info{id=SessionId}] = ets:lookup(Ets,Pid),
    SessionId.
  
  
  gen_id() ->
    crypto:rand_uniform(0,9007199254740992).
  
  publish_metaevent(_,_,_,_,#state{meta_events=disabled}) ->
    ok;
  publish_metaevent(Event,TopicUri,SessionId,SecondArg,#state{ets=Ets}) ->
    case binary:part(TopicUri,{1,5}) == <<"wamp.">> of
      true ->
        % do not fire metaevents on wamp. uris
        ok;
      false ->
        MetaTopic = case Event of
                      on_create -> <<"wamp.subscription.on_create">>;
                      on_subscribe -> <<"wamp.subscription.on_subscribe">>;
                      on_unsubscribe -> <<"wamp.subscription.on_unsubscribe">>;
                      on_delete -> <<"wamp.subscription.on_delete">>
                    end,
        {ok,_} = publish(MetaTopic,#{},[SessionId,SecondArg],undefined,no_session,#data{ets=Ets})
    end,
    ok.
  
  
  