Re: Fwd: StandaloneProcessor

  • From: Dean Pehrsson-Chapman <dean@...>
  • To: users@...
  • Subject: Re: Fwd: StandaloneProcessor
  • Date: Wed, 7 Aug 2013 17:52:22 +0100

Much appreciated.


On 7 August 2013 17:47, Oleksiy Stashok <oleksiy.stashok@...> wrote:

>  Hi Dean,
>
>
>
> On 07.08.13 01:27, Dean Pehrsson-Chapman wrote:
>
> Thanks - this does make sense.
>
>  Coming full circle, I can make this approach work with the new
> connection pool implementation if I use a Map (keyed on Connection) of
> BlockQueues, does that sound right?
>
> I'd recommend to use Connection Attributes instead of single Map. Like:
>
>     private static final Attribute<BlockingQueue<Object>>
> RESPONSE_QUEUE_ATTR = Attribute.create("response-queue", new
> NullaryFunction<Object>() {
>
>             @Override
>             public Object evaluate() {
>                    return new LinkedBlockingQueue(); // you can use
> different blocking queue implementation
>             }
>         }
>     );
>
> ................
>
>
>             public Client() {
>                     final FilterChain clientFilterChain =
> FilterChainBuilder.stateless()
>                             .add(new TransportFilter())
>                             .add(new HttpClientFilter())
>                             .add(new BaseFilter() {
>                         @Override
>                         public NextAction handleRead(FilterChainContext
> ctx) throws IOException {
>
> RESPONSE_QUEUE_ATTR.get(ctx.getConnection()).add(EOF_PACKET);
>                             return ctx.getStopAction();
>                         }
>
>                         @Override
>                         public NextAction handleClose(FilterChainContext
> ctx) throws IOException {
>
> RESPONSE_QUEUE_ATTR.get(ctx.getConnection()).add(EOF_PACKET);
>
>                             return ctx.getStopAction();
>                         }
>                     }).build();
>
>                     transport = TCPNIOTransportBuilder.newInstance()
>                             .setProcessor(clientFilterChain)
>                             .build();
>             }
>
> ................
>
>             public HttpPacket read() throws IOException {
>                     synchronized (readSync) {
>                           if (!isClosed) {
>                               final Object packet =
> RESPONSE_QUEUE_ATTR.get(ctx.getConnection()).take();
>
>                               if (packet != EOF_PACKET) {
>                                      return (HttpPacket) packet;
>                               }
>
>                               isClosed = true;
>                           }
>
>                           throw new EOFException("The connection is
> closed");
>                     }
>             }
>
> Thanks.
>
> WBR,
> Alexey.
>
>
>  Cheers,
> Dean
>
>
> On 6 August 2013 22:18, Oleksiy Stashok <oleksiy.stashok@...>wrote:
>
>>  Hi Dean,
>>
>> I see what you mean.
>> In general we suggest to not use standalone mode and it will be removed
>> in Grizzly 3.
>> We recommend to use FilterChain approach for the server- and the
>> client-side code.
>>
>> With the server-side it's clear, right?
>> But with the client-side code we used to have entire logic to be
>> synchronous, like with Socket or HttpURLConnection, you send a request and
>> want to read a response right away in the same thread. It's still possible
>> to achieve this kind of behavior with Grizzly FilterChain.
>>
>> Here is a simple HTTP client example (not the real code):
>>
>> public class Client {
>>             private static final Object EOF_PACKET = new Object();
>>
>>             private final TCPNIOTransport transport;
>>             private final BlockingQueue<Object> resultQueue = new
>> LinkedTransferQueue<Object>();
>>
>>             private final Object readSync = new Object();
>>
>>             private boolean isClosed;
>>             private Connection connection;
>>
>>             public Client() {
>>                     final FilterChain clientFilterChain =
>> FilterChainBuilder.stateless()
>>                             .add(new TransportFilter())
>>                             .add(new HttpClientFilter())
>>                             .add(new BaseFilter() {
>>                         @Override
>>                         public NextAction handleRead(FilterChainContext
>> ctx) throws IOException {
>>                             resultQueue.add((HttpPacket)
>> ctx.getMessage());
>>                             return ctx.getStopAction();
>>                         }
>>
>>                         @Override
>>                         public NextAction handleClose(FilterChainContext
>> ctx) throws IOException {
>>                             resultQueue.add(EOF_PACKET);
>>                             return ctx.getStopAction();
>>                         }
>>                     }).build();
>>
>>                     transport = TCPNIOTransportBuilder.newInstance()
>>                             .setProcessor(clientFilterChain)
>>                             .build();
>>             }
>>
>>             public synchronized void connect(SocketAddress dstAddress)
>> throws IOException {
>>                     connection = transport.connect(dstAddress).get(10,
>> TimeUnit.SECONDS);
>>             }
>>
>>             public void write(HttpPacket httpPacket) {
>>                     connection.write(httpPacket);
>>             }
>>
>>             public HttpPacket read() throws IOException {
>>                     synchronized (readSync) {
>>                           if (!isClosed) {
>>                               final Object packet = resultQueue.take();
>>                               if (packet != EOF_PACKET) {
>>                                      return (HttpPacket) packet;
>>                               }
>>
>>                               isClosed = true;
>>                           }
>>
>>                           throw new EOFException("The connection is
>> closed");
>>                     }
>>             }
>> }
>>
>> As I said it's not a complete code, but it can give you an idea how the
>> FilterChain-based Client may look like.
>>
>> Hope that will help.
>>
>> WBR,
>> Alexey.
>>
>>
>> On 06.08.13 12:56, Dean Pehrsson-Chapman wrote:
>>
>>  Sure, I'm just interacting with an echo server (which is working fine)
>> in a synchronous way.
>>
>>       // Create TCP transport
>>         final TCPNIOTransport transport =
>>                 TCPNIOTransportBuilder.newInstance().build();
>>
>>          transport.setProcessor(FilterChainBuilder.stateless().build());
>>         transport.configureStandalone(true);
>>         transport.start();
>>
>>              Connection c = transport.connect(HOST, PORT).get();
>>              c.write(HeapBuffer.wrap("hello".getBytes()));
>>             ReadResult r = (ReadResult) c.read().get();
>>
>>  FilterChainContext.read() would work fine, but that means my transport,
>> once started, can only do a particular set of things (the things defined in
>> the filter).  I am trying to bridge a legacy communication system into
>> grizzly.  The server part works fine, but I think I may be trying to push a
>> square peg into a round hole with the client.
>>
>>  Cheers,
>> Dean
>>
>>
>> On 6 August 2013 15:47, Oleksiy Stashok <oleksiy.stashok@...>wrote:
>>
>>>  Hi Dean,
>>>
>>> can you pls. share your code or attach a simple example of what you're
>>> trying to achieve.
>>> Did you try FilterChainContext.read()?
>>>
>>> Thanks.
>>>
>>> WBR,
>>> Alexey.
>>>
>>>
>>> On 06.08.13 06:42, Dean Pehrsson-Chapman wrote:
>>>
>>> I tend to get a lot of
>>>
>>>  java.lang.NullPointerException
>>> at org.glassfish.grizzly.asyncqueue.AsyncReadQueueRecord.isFinished
>>>
>>>  I see a previous user who hit this issue was advised to use the filter
>>> method.   My particular use case is to use the shiny new connection pool -
>>> how can a client grab a connection and do some work with a filter?  I 
>>> don't
>>> understand the thinking behind the architecture here.
>>>
>>>  Any help gratefully received.
>>>
>>>  Cheers,
>>> Dean
>>>
>>>
>>> ---------- Forwarded message ----------
>>> From: Dean Pehrsson-Chapman <dean@...>
>>> Date: 6 August 2013 10:16
>>> Subject: StandaloneProcessor
>>> To: users@...
>>>
>>>
>>> Hi,
>>>
>>>  After some fiddling about I've realised that synchronously using
>>>
>>>  connection.read()
>>>
>>>  doesn't work with a filter chain - the message is handled by the
>>> filter chain and not reported to connection.read.  Setting the processor 
>>> to
>>> be a StandaloneProcessor works fine, but that seems odd to me as now you
>>> can't take advantage of the filterchain mechanism.  Is it correct to say
>>> that there are incompatible ways of doing sync client (or even async - a
>>> completion handler will never be called either) and async server?
>>>
>>>  Or am I just doing it wrong?
>>>
>>>  Cheers,
>>> Dean
>>>
>>>
>>>
>>
>>
>
>


StandaloneProcessor

Dean Pehrsson-Chapman 08/06/2013

Fwd: StandaloneProcessor

Dean Pehrsson-Chapman 08/06/2013

Re: Fwd: StandaloneProcessor

Oleksiy Stashok 08/06/2013

Re: Fwd: StandaloneProcessor

Dean Pehrsson-Chapman 08/06/2013

Re: Fwd: StandaloneProcessor

Oleksiy Stashok 08/06/2013

Re: Fwd: StandaloneProcessor

Dean Pehrsson-Chapman 08/07/2013

Re: Fwd: StandaloneProcessor

Oleksiy Stashok 08/07/2013

Re: Fwd: StandaloneProcessor

Dean Pehrsson-Chapman 08/07/2013
Terms of Use; Privacy Policy; Copyright ©2013-2015 (revision 20150626.29986a4)
 
 
Close
loading
Please Confirm
Close