handle product update notice

src/kvs_feed.erl

@@ -129,39 +129,43 @@ handle_notice([kvs_feed, _, Owner, entry, Eid, add],
               [#entry{feed_id=Fid}=Entry|_],
               #state{owner=Owner} = S) ->
     case lists:keyfind(Fid,2, S#state.feeds) of false -> skip;
-      {_,_} ->
+    {_,_} ->
-        error_logger:info_msg("kvs_feed => Entry ~p added to feed ~p.", [Eid, Fid]),
+        error_logger:info_msg("[kvs_feed] => Add entry ~p to feed ~p.", [Eid, Fid]),
         E = Entry#entry{id = {Eid, Fid}, entry_id = Eid, feeds=[comments]},
-        kvs:add(E) end,
+        kvs:add(E),
+        msg:notify([kvs_feed, entry, {Eid, Fid}, added], [E]) end,
     {noreply, S};
 
-handle_notice([kvs_feed,_, Owner, entry, {_, Fid}, edit],
+handle_notice([kvs_feed,_, Owner, entry, {Eid, FeedName}, edit],
-              [#entry{entry_id=Eid}=Entry|_],
+              [#entry{}=Entry],
               #state{owner=Owner, feeds=Feeds}=S) ->
+    case lists:keyfind(FeedName,1,Feeds) of false -> skip;
+    {_,Fid}-> case kvs:get(entry, {Eid, Fid}) of {error,_}-> skip;
+        {ok, E} ->
+            error_logger:info_msg("[kvs_feed] => Update entry ~p in feed ~p", [Eid, Fid]),
+            Upd = E#entry{description=Entry#entry.description,
+                        title = Entry#entry.title,
+                        media = Entry#entry.media,
+                        etc   = Entry#entry.etc,
+                        type  = Entry#entry.type},
+            kvs:put(Upd),
+            msg:notify([kvs_feed, entry, {Eid, Fid}, updated], [Upd]) end end,
+    {noreply, S};
 
-  case lists:keyfind(Fid,2,Feeds) of false -> skip;
+handle_notice([kvs_feed,_, Owner, entry, {_,Fid}=Id, delete],
-    {_,_} -> case kvs:get(entry, {Eid, Fid}) of {error, not_found}-> skip; 
+              [],
-        {ok, E} -> 
+              #state{owner=Owner, feeds=Feeds} = State) ->
-          error_logger:info_msg("kvs_feed => Entry ~p updated in feed ~p", [Eid, Fid]),
+    case lists:keyfind(Fid,2,Feeds) of false -> skip;
-          kvs:put(E#entry{description=Entry#entry.description,
+    _ ->
-                          title = Entry#entry.title,
+        error_logger:info_msg("[kvs_feed] => Remove entry ~p from feed ~p", [Id, Fid]),
-                          media = Entry#entry.media,
+        kvs:remove(entry, Id),
-                          etc   = Entry#entry.etc,
+        msg:notify([kvs_feed, entry, Id, deleted], []) end,
-                          type  = Entry#entry.type}) end end,
-
-  {noreply, S};
-
-handle_notice([kvs_feed,_, Owner, entry, Fid, delete],
-              [#entry{id=Id}|_], #state{owner=Owner, feeds=Feeds} = State) ->
-    error_logger:info_msg("Delete notice: ~p ", [Owner]),
-  case lists:keyfind(Fid,2,Feeds) of false -> skip;
-    {_,_} -> error_logger:info_msg("kvs_feed => Remove entry ~p from feed ~p", [Id, Fid]), kvs:remove(entry, Id) end,
   {noreply, State};
 
 handle_notice([kvs_feed,_,Owner,comment,_,add],
-              [#comment{entry_id={_,EFid}}=C,_,_],
+              [#comment{entry_id={_,Fid}}=C,_,_],
               #state{owner=Owner, feeds=Feeds} = S) ->
-    case lists:keyfind(EFid,2,Feeds) of false -> skip; {_,_}-> kvs:add(C) end,
+    case lists:keyfind(Fid,2,Feeds) of false -> skip; {_,_}-> kvs:add(C) end,
     {noreply, S};
 
 handle_notice(["kvs_feed","likes", _, _, "add_like"] = Route,  % _, _ is here beacause of the same message used for comet update

src/kvs_group.erl

@@ -3,6 +3,7 @@
 -include_lib("kvs/include/kvs.hrl").
 -include_lib("kvs/include/users.hrl").
 -include_lib("kvs/include/groups.hrl").
+-include_lib("kvs/include/feeds.hrl").
 -include_lib("kvs/include/accounts.hrl").
 -include_lib("kvs/include/feed_state.hrl").
 -include_lib("kvs/include/config.hrl").
@@ -133,18 +134,26 @@ handle_notice(["kvs_group", "remove", GroupName] = Route,
     delete(GroupName),
     {noreply, State};
 
-handle_notice([kvs_group, join, GroupName],
+handle_notice([kvs_group, join, Owner],
-    {UserName, Type}, #state{type=Type} = State) ->
+              [{FeedName, Who}, Entry],
-  error_logger:info_msg("Join group:  ~p State type:~p", [GroupName, Type]),
+              #state{owner = Owner, feeds=Feeds} = State) ->
-  join(UserName, GroupName),
+    error_logger:info_msg("[kvs_group] ~p Join group:  ~p", [Who, Owner]),
+    join(Who, Owner),
+    case lists:keyfind(FeedName, 1, Feeds) of false -> skip;
+    {_,Fid} -> msg:notify([kvs_feed, group, Owner, entry, Who, add],
+                          [Entry#entry{id={Who, Fid},feed_id=Fid,to={group, Owner}}, {}, Fid]) end,
+
 %    subscription_mq(group, add, UserName, GroupName),
-  {noreply, State};
+    {noreply, State};
+
+handle_notice([kvs_group, leave, Owner],
+              [{FeedName, Who}],
+              #state{owner = Owner, feeds=Feeds} = State) ->
+    error_logger:info_msg("[kvs_group] ~p leave group ~p", [Who, Owner]),
+    leave(Who, Owner),
+    case lists:keyfind(FeedName, 1, Feeds) of false -> skip;
+    {_,Fid} -> msg:notify([kvs_feed, group, Owner, entry, {Who, Fid}, delete], []) end,
 
-handle_notice(["kvs_group", "leave", GroupName] = Route,
-    Message, #state{owner = Owner, type =Type} = State) ->
-    error_logger:info_msg("queue_action(~p): remove_from_group: Owner=~p, Route=~p, Message=~p", [self(), {Type, Owner}, Route, Message]),
-    {UserName} = Message,
-    leave(UserName,GroupName),
 %    subscription_mq(group, remove, UserName, GroupName),
     {noreply, State};
 

src/kvs_product.erl

@@ -28,25 +28,28 @@ delete(Name) ->
 
 subscription_mq(Type, Action, Who, Whom) ->
     case mqs:open([]) of
-        {ok,Channel} ->
+    {ok,Channel} ->
-            case {Type,Action} of 
+        case {Type,Action} of
-                {user,add}     -> mqs_channel:bind_exchange(Channel, ?USER_EXCHANGE(Who), ?NOTIFICATIONS_EX, rk_product_feed(Whom));
+        {user,add}->
-                {user,remove}  -> mqs_channel:unbind_exchange(Channel, ?USER_EXCHANGE(Who), ?NOTIFICATIONS_EX, rk_product_feed(Whom)) end,
+            mqs_channel:bind_exchange(Channel, ?USER_EXCHANGE(Who), ?NOTIFICATIONS_EX, rk_product_feed(Whom));
-            mqs_channel:close(Channel);
+        {user,remove} ->
-        {error,Reason} -> error_logger:info_msg("subscription_mq error: ~p",[Reason]) end.
+            mqs_channel:unbind_exchange(Channel, ?USER_EXCHANGE(Who), ?NOTIFICATIONS_EX, rk_product_feed(Whom)) end,
+        mqs_channel:close(Channel);
+    {error,Reason} -> error_logger:info_msg("subscription_mq error: ~p",[Reason]) end.
 
 init_mq(Product=#product{}) ->
     Groups = kvs_group:participate(Product),
     ProductExchange = ?USER_EXCHANGE(Product#product.id),
     ExchangeOptions = [{type, <<"fanout">>}, durable, {auto_delete, false}],
     case mqs:open([]) of
-        {ok, Channel} ->
+    {ok, Channel} ->
-            error_logger:info_msg("Cration Exchange: ~p,",[{Channel,ProductExchange,ExchangeOptions}]),
+        error_logger:info_msg("Cration Exchange: ~p,",[{Channel,ProductExchange,ExchangeOptions}]),
-            mqs_channel:create_exchange(Channel, ProductExchange, ExchangeOptions),
+        mqs_channel:create_exchange(Channel, ProductExchange, ExchangeOptions),
-            Relations = build_product_relations(Product, Groups),
+        Relations = build_product_relations(Product, Groups),
-            [ mqs_channel:bind_exchange(Channel, ?USER_EXCHANGE(Product#product.id), ?NOTIFICATIONS_EX, Route) || Route <- Relations],
+        [ mqs_channel:bind_exchange(Channel, ?USER_EXCHANGE(Product#product.id), ?NOTIFICATIONS_EX, Route)
-            mqs_channel:close(Channel);
+            || Route <- Relations],
-        {error,Reason} -> error_logger:info_msg("init_mq error: ~p",[Reason]) end.
+        mqs_channel:close(Channel);
+    {error,Reason} -> error_logger:info_msg("init_mq error: ~p",[Reason]) end.
 
 build_product_relations(Product, Groups) -> [
     mqs:key( [kvs_product, '*', Product]),
@@ -99,6 +102,51 @@ handle_notice(["kvs_product", "update", _Who],
     kvs:put(NewProduct),
     {noreply, State};
 
+handle_notice([kvs_product, Owner, update],
+              [#product{}=Product, Recipients, Is, Fs], #state{owner=Owner} = State) ->
+    error_logger:info_msg("[kvs_product] Update product ~p", [Owner]),
+    case kvs:get(product, Owner) of {error,E}->
+        msg:notify([kvs_product, product, Owner, updated], [{error,E}, Is, Fs]);
+    {ok, #product{}=P} ->
+        Id = P#product.id,
+        UpdProduct = P#product{
+            title = Product#product.title,
+            brief = Product#product.brief,
+            cover = Product#product.cover,
+            price = Product#product.price,
+            currency = Product#product.currency},
+        kvs:put(UpdProduct),
+
+        Medias = case Product#product.cover of undefined -> [];
+            File -> [#media{url=File,
+                thumbnail_url = filename:join([filename:dirname(File),"thumbnail",filename:basename(File)]) }] end,
+
+        Entry = #entry{
+            created=P#product.created,
+            entry_id=Id,
+            from=P#product.owner,
+            type= product,
+            media=Medias,
+            title=Product#product.title,
+            description=Product#product.brief,
+            shared=""},
+
+        Groups = ordsets:from_list([Gid || {Type,Gid} <- Recipients, Type==group]),
+        Participate = ordsets:from_list([Gid || #group_subscription{where=Gid} <- kvs_group:participate(Id)]),
+        Intersection = ordsets:intersection(Groups, Participate),
+        Leave = ordsets:subtract(Participate, Intersection),
+        Join = ordsets:subtract(Groups, Intersection),
+
+        [msg:notify([kvs_group, leave, Gid], [{products, Id}]) || Gid <- Leave],
+        [msg:notify([kvs_group, join, Gid], [{products, Id}, Entry]) || Gid <- Join],
+
+        [msg:notify([kvs_feed, RouteType, To, entry, {Id, products}, edit], [Entry])
+            || {RouteType, To} <- [{user, P#product.owner} | [{group, G} || G <- Intersection]]],
+        msg:notify([kvs_product, product, Owner, updated], [UpdProduct, Is])
+    end,
+    {noreply, State};
+
+
 handle_notice(_Route, _Message, State) ->
   %error_logger:info_msg("Unknown USERS notice"),
   {noreply, State}.