We followed a model similar to the first one Robbie outlined below.
- We dedicate a thread to running the Reactor.
- When an application does a ‘send’ the message is put on an internal queue and
reactor.wakeup() is used just in case that thread is asleep.
- We attempt physical message transmission whenever the reactor quiesces or
credits appear on a link.
- We periodically sweep for messages that are stuck in the transmission
queue(s) and reject them back to the application via an onError() handler if
they’ve been pending for some period of time. I think this should be
best-practice anyway but is more important in a broker-less/dispatch-router
based setup.
- On the receiving side you need to decide if you’re willing to let your
application block the reactor thread or decouple it via another internal queue
– what you’ll need to make a call on is when to change the Delivery disposition
to mark it as accepted.
- We created a simplistic mechanism to dispatch commands for execution on the
Reactor thread that we use for various housekeep activities.
—
Jem
Principal Architect, Core Payments
On 7/19/16, 2:36 AM, "Robbie Gemmell" <robbie.gemm...@gmail.com> wrote:
From previous discussion on the subject I believe there are two main
options for interacting with other threads in such manner:
- Use the 'wakeup' functionality from the Selectable used for the
timer, calling reactor.wakeup() to wake the reactor thread from its
selector.select() call if needed, then doing any desired processing in
the onReactorQuiesced callback fired before it goes back into select
again either for the configured timeout or until a timer task needs
processing.
- Write your own 'Selectable' and do the desired processing in its
callbacks.
Robbie
On 18 July 2016 at 20:52, Garlapati Sreeram Kumar <sreer...@live.com> wrote:
> Hello All!
>
> We are using Proton-j Reactor framework & are trying to offer Send API in
our EventHub Client offering to talk to Microsoft Azure Event Hubs
(EventSender.Send(eventData)).
>
> To simplify the scenario lets say, we mapped each EventSender (in our
client) to 1 Reactor – which maps to 1 Connection – which has 1 Session –
which has only 1 SenderLink. We run Reactor on 1 dedicated Thread. Users of our
EventHubs client Framework can call Send(..) concurrently from different
threads.
>
> We are trying to see what’s the best way to implement this using the
current Reactor Artifacts.
>
> - Given that, the Reactor implementation assumes
SenderLink.send() happens on the Same thread as that of Reactor – what was the
artifact intended in Reactor’s Design, to be used – to implement Send? To
explain more – ‘onReceive’ is naturally a reactive event – but Send is an
Action which could start from a completely different thread – so, how should
Send tap into Reactor’s event-delegation model?
>
> Here’s the challenge we are facing: Lets imagine the below events in
chronological order:
>
> 1. User created a SenderLink
>
> 2. user started 10 parallel sends – our Client SDK parked all of
this sends in a SendWaiter Queue – so that we will invoke the SenderLink.send()
based on a Reactor Event (so that they run on the same thread)
>
> 3. Service responded with Link Attach frame & issued a Credit of 300
>
> 4. Based on this event our SDK created 10 new Delivery’s and
completed all 10 sends on the Reactor Thread.
>
> 5. Now, if the User thread invokes more Sends – what are the events
on which we can invoke these sends – so that they can run on the Same thread of
Reactor ? Currently we implemented a SendPump – using Timer.schedule – every
50ms – feed off our SendWaiter Queue and invoke SenderLink.send() on the
Reactor Thread.
>
> Looks like, using ‘Selectable’ could be a way – but, wanted to
double-check the direction…
>
> 'Timer approach’ currently works, but doesn’t look like the right way &
hence the probe…
>
> Thanks in Advance!
> Sree
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscr...@qpid.apache.org
For additional commands, e-mail: users-h...@qpid.apache.org
