<div dir="ltr">If we remove all the unrelated implementation details then we see a picture:<div><br></div><div>handle_info({incoming, Id, SomeDetails}, State) -></div><div> start_processing(SomeDetails),</div><div> {noreply, State};</div><div>handle_info({processed, Result}, State) -></div><div> Id = ???, % how to get the value</div><div> finalise(Id, Result),</div><div> {noreply, State}<br><div><br></div><div>I can suggest the following approaches to do that assuming you cannot force the processor to accept and pass your value back.</div><div>1. If you can switch to synchronous processing (which may be desirable to prevent overloading of the processor) you can store the Id in the process State.</div><div> handle_info({incoming, Id, SomeDetails}, #{current_id := undefined} = State) -></div><div> start_processing(SomeDetails),</div><div> {noreply, State#{current_id => Id}};</div><div> handle_info({processed, Result}, #{current_id := Id} = State) -></div><div> finalise(Id, Result),</div><div> {noreply, State#{current_id => undefined}}</div><div><br></div><div>2. If the processing must be asynchronous and there may be multiple inflight processings, you can:</div><div> a) make start_processing to accept the Id and to spawn a new process which would receive the processed message (tcp reponse)</div><div> handle_info({incoming, Id, SomeDetails}, State) -></div><div> spawn_link(fun -></div><div> start_processing(SomeDetails),</div><div> receive</div><div> {processed, Result} -></div><div> finalise(Id, Result)</div><div> end</div><div> end),</div><div> {noreply, State}</div><div><br></div><div>3. make mapping from sockets to Ids and keep it in the state</div><div> handle_info({incoming, Id, SomeDetails}, State) -></div><div> Socket = start_processing(SomeDetails),</div><div> {noreply, State#{Socket => Id}};</div><div> handle_info({processed, Socket, Result}, State) -></div><div> #{Socket := Id} = State,</div><div> finalise(Id, Result),</div><div> {noreply, maps:delete(State, Socket)}.</div><div><br></div><div><br clear="all"><div><div dir="ltr" class="gmail_signature" data-smartmail="gmail_signature"><div dir="ltr">Kind regards,<div>Dmitry Belyaev</div></div></div></div><br></div></div></div><br><div class="gmail_quote"><div dir="ltr" class="gmail_attr">On Fri, Aug 16, 2019 at 11:51 PM Sébastien BRICE <<a href="mailto:otb@opentelecom.fr">otb@opentelecom.fr</a>> wrote:<br></div><blockquote class="gmail_quote" style="margin:0px 0px 0px 0.8ex;border-left:1px solid rgb(204,204,204);padding-left:1ex">Hello there,<br>
<br>
I am a bit new to the Erlang Environment<br>
<br>
I am writing an emailtesting application that filters incoming email <br>
with a randomly generated routing_keys on a topic exchange to make <br>
emails entering my system<br>
<br>
Once they are delivered (and processed) on an queue, I want to label <br>
them again with the previously randomly routing_key to route them to <br>
another exchange to make them ready for the final consume.<br>
<br>
This 2nd producing step is causing me real troubles<br>
<br>
I am getting data back from a tcp socket (processed by a third-tier <br>
program: spamassassin) with handle_info pattern matching<br>
<br>
I rely on a gen_server to consume messages first through the regular <br>
amqp_client/include/amqp_client.hrl Library<br>
<br>
I use handle_info in my gen_server behaviour and then pattern match on <br>
the parameters.<br>
<br>
Detecting delivered AMQP message is done through function heads <br>
(records) in handle_info callback<br>
<br>
***My gen_server****<br>
handle_info({#'basic.deliver'{routing_key=Key, consumer_tag=Tag}, <br>
Content}, State) -><br>
#amqp_msg{props = Properties, payload = Payload} = Content,<br>
#'P_basic'{message_id = MessageId, headers = Headers} = Properties,<br>
send_to_spamassassin:calcule_score(Payload),<br>
{noreply, State};<br>
handle_info(Msg, State) -><br>
case Msg of<br>
{_,_,Data} -><br>
scored_email:main(Data);<br>
{_,_} -><br>
end,<br>
{noreply, State}.<br>
<br>
***send_to_spamassassin function ***<br>
calcule_score(Message) -><br>
case gen_tcp:connect("localhost", 783, [{mode, binary}]) of<br>
{ok, Sock} -><br>
…<br>
gen_tcp:send(Sock, Message2);<br>
{error,_} -><br>
io:fwrite("Connection error! Quitting...~n")<br>
end.<br>
<br>
<br>
TCP socket is nice to talk with spamassassin, it returns me a 3-tuple <br>
with binary string data like that:<br>
<br>
{tcp,#Port<0.55>,<<"SPAMD/1.1 0 EX_OK\r\nContent-length: 564\r\nSpam: <br>
True ; 7.9 / 5.0\r\n\r\nReceived: from localhost by <br>
<a href="http://XXXX.ikexpress.com" rel="noreferrer" target="_blank">XXXX.ikexpress.com</a>\n\twith SpamAssassin (version 3.4.2);\n\tThu, 15 Aug <br>
2019 21:44:12 +0200\nX-Spam-Checker-Version: SpamAssassin 3.4.2 <br>
(2018-09-13) on\n\<a href="http://tXXXXX.ikexpress.com" rel="noreferrer" target="_blank">tXXXXX.ikexpress.com</a>\nX-Spam-Flag: YES\nX-Spam-Level: <br>
*******\nX-Spam-Status: Yes, score=7.9 required=5.0 <br>
tests=EMPTY_MESSAGE,MISSING_DATE,\n\tMISSING_FROM,MISSING_HEADERS,MISSING_MID,MISSING_SUBJECT,\n\tNO_HEADERS_MESSAGE,NO_RECEIVED,NO_RELAYS <br>
autolearn=no\n\tautolearn_force=no version=3.4.2\nMIME-Version: <br>
1.0\nContent-Type: multipart/mixed; <br>
boundary=\"----------=_5D55B60C.D2FC2670\"\n\n">>}<br>
<br>
The loop in the second handle_info match OK the answer from the <br>
listening gen_tcp server, but I have to do the packaging to send it to a <br>
topic Exchange (topic_scored_email exchange)<br>
<br>
***scored_email***<br>
main(Argv) -><br>
{ok, Connection} = <br>
amqp_connection:start(#amqp_params_network{virtual_host = <<"/">>}),<br>
{ok, Channel} = amqp_connection:open_channel(Connection),<br>
amqp_channel:call(Channel, #'exchange.declare'{exchange = <br>
<<"topic_scored_email">>,type = <<"topic">>}),<br>
{RoutingKey, Message} = case Argv of<br>
…<br>
%DOING PATTERN MATCHING THAT WORKS HERE<br>
…<br>
end,<br>
amqp_channel:cast(Channel,#'basic.publish'{exchange = <br>
<<"topic_scored_email">>,routing_key = RoutingKey},#amqp_msg{payload = <br>
Message}),<br>
<br>
First issue is type of the data (binary string) but I guess it can be <br>
workarounded using BIF binary_to_tuple or stuff like that.<br>
<br>
What I struggle to understand is how I could pass the righ RoutingKey, <br>
since Erlang is functionnal, there is no side effect or assignation.<br>
<br>
That change in format data (AMQP --> raw tcp --> then AMQP again) seems <br>
impossible (to me) to achieve with OTP abstraction<br>
<br>
However I would like to reassemble every processed message with the <br>
right routing key matched 5 lines above.<br>
<br>
How could I modify my code, to do that ? I come from imperative language <br>
and reach my limit here…<br>
<br>
Yours<br>
<br>
PS I know it is more a rabbitmq issue and I might be more successful to <br>
post on stackoverflow or rabbitmq google groups but I feel <br>
#Erlang-questions could come handy on that topic<br>
<br>
_______________________________________________<br>
erlang-questions mailing list<br>
<a href="mailto:erlang-questions@erlang.org" target="_blank">erlang-questions@erlang.org</a><br>
<a href="http://erlang.org/mailman/listinfo/erlang-questions" rel="noreferrer" target="_blank">http://erlang.org/mailman/listinfo/erlang-questions</a><br>
</blockquote></div>