MessageToMessage Encoder/Decoder for AddressedEnvelope

[madde...@gmail.com]

Hi,

I'd like to use a MessageToMessageDecoder to transform from one type to another while sending a message from a client to a remote server.  The MessageToMessage Encoder/Decoder allow this to be accomplished easily.  When using UDP and an AddressedEnvelope to preserve remote address info, this is not the case.  

The following snippet shows a simplified channel pipeline to hopefully illustrate my use case.

ChannelInitializer<DatagramChannel> datagramChannelInitializer = new ChannelInitializer<DatagramChannel>() {

                    @Override

                    public void initChannel(DatagramChannel ch) throws Exception {

                                ch.pipeline().addLast(new DatagramPacketDecoder(new Type1Decoder()))

                                .addLast(new Type2MessageToMessageDecoder())

                                .addLast(new Type2MessageToMessageEncoder())

                                .addLast(new DatagramPacketEncoder<>(new Type1Encoder()));
                    }
}

During encoding, I wrap my type1 payload in an AddressedEnvelope to specify the remote address.  The Type2MessageToMessageEncoder is never invoked because AddressedEnvelope<Type1, SocketAddress> does not match Type2.  I tried to create my own class AddressedEnvelopeMessageToMessageEncoder similar to the DatagramPacketEncoder which delegates to a provided encoder but the encode method is not visible.

What is the recommended approach for this kind of use case?  Any help appreciated.

Thanks,

Joe

[Norman Maurer]

Can you show me the code that not worked ?

--
You received this message because you are subscribed to the Google Groups "Netty discussions" group.
To unsubscribe from this group and stop receiving emails from it, send an email to netty+un...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/netty/788b5d49-63fa-4dff-b522-81d90118ada3%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Norman Maurer]

So all good?

Am 18.08.2016 um 11:22 schrieb Joseph Madden <madde...@gmail.com>:

Hi Norman,

Apologies for not getting back to you sooner.  My initial attempt at this had a bug in it.  I had the encoders in the wrong order.  I ended up with something like this:

ChannelInitializer<DatagramChannel> datagramChannelInitializer = new ChannelInitializer<DatagramChannel>() {
                    @Override
                    public void initChannel(DatagramChannel ch) throws Exception {

                                ch.pipeline().addLast(new DatagramPacketDecoder(new Type1Decoder()))
                                .addLast(new Type2MessageToMessageDecoder())

.addLast(new DatagramPacketEncoder<>(new Type1Encoder())
                                .addLast(new Type2MessageToMessageEncoder())
                                );
                    }
}

On Thursday, 4 August 2016 14:13:19 UTC+1, Norman Maurer wrote:

Can you show me the code that not worked ?

On 02 Aug 2016, at 15:06, madde...@gmail.com wrote:

Hi,

I'd like to use a MessageToMessageDecoder to transform from one type to another while sending a message from a client to a remote server.  The MessageToMessage Encoder/Decoder allow this to be accomplished easily.  When using UDP and an AddressedEnvelope to preserve remote address info, this is not the case.  

The following snippet shows a simplified channel pipeline to hopefully illustrate my use case.

ChannelInitializer<DatagramChannel> datagramChannelInitializer = new ChannelInitializer<DatagramChannel>() {

                    @Override

                    public void initChannel(DatagramChannel ch) throws Exception {

                                ch.pipeline().addLast(new DatagramPacketDecoder(new Type1Decoder()))

                                .addLast(new Type2MessageToMessageDecoder())

--

You received this message because you are subscribed to the Google Groups "Netty discussions" group.
To unsubscribe from this group and stop receiving emails from it, send an email to netty+un...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/netty/78f3bafd-ba4b-4b97-8cea-0725e89a3c6d%40googlegroups.com.

[Joseph Madden]

Hi Norman,

Apologies for the late response.  The encoders in my example above were in the wrong order which prevented the handlers from being invoked.

To briefly summarise my use case :

Type2 -> Type1 -> Outbound

I ended up with something similar to the following code snippets:

ChannelInitializer<DatagramChannel> datagramChannelInitializer = new ChannelInitializer<DatagramChannel>() {
                    @Override
                    public void initChannel(DatagramChannel ch) throws Exception {


                                ch.pipeline().addLast(new DatagramPacketDecoder(new Type1Decoder()))
                                .addLast(new Type2MessageToMessageDecoder())

                                .addLast(new DatagramPacketEncoder<>(new Type1Encoder())
                                .addLast(new AddressEnvelopeType2MessageToMessageEncoder()));
                    }
}


   
public class UdpClient{


 private final SocketAddress socketAddress
 private Bootstrap bootstrap;
 private Channel channel;


 public UdpClient(SocketAddress socketAddress)
 {
 this.socketAddress = socketAddress;
 this.bootstrap = new Bootstrap();
 this.bootstrap.handler(channelInitializer);
 }


    public void write(Type2 message, boolean await) {
        AddressedEnvelope<Type2, SocketAddress> addressedEnvelope = new DefaultAddressedEnvelope<Type2, SocketAddress>(message,
                socketAddress, channel.localAddress());
        if (await) {
            channel.writeAndFlush(addressedEnvelope).awaitUninterruptibly();
        } else {
            channel.writeAndFlush(addressedEnvelope, channel.voidPromise());
        }
    }
}


public class AddressEnvelopeType2MessageToMessageEncoder
        extends MessageToMessageEncoder<AddressedEnvelope<Type2, SocketAddress>> {


    @Override
    protected void encode(ChannelHandlerContext ctx, AddressedEnvelope<Type2Message, SocketAddress> envelope,
            List<Object> out) throws Exception {
        out.add(new DefaultAddressedEnvelope<Type1, SocketAddress>(
                new Type1(envelope.content().getParameter()),
                envelope.recipient(), envelope.sender()));
    }


}

The UdpClient class is instantiated with the socketAddress of the remote server.  If I'd like to reuse the Netty AddressedEnvelope for addressing purposes and also transform the message using a MessageToMessageEncoder/Decoder in the channel pipeline, I have to create a MessageToMessageEncoder similar to the one above.

Is there a better way of doing this?

Since it's possible to put multiple MessageToMessage encoder/decoders on the same channel pipeline It would be nice to be able to preserve the Address information along the pipeline for the UDP use case.  I attempted to create a generic MessageToMessageEncoder that would do this similar in nature to the DatagramPacketEncoder which would delegate to a provided encoder.  I didn't have alot of time in the end so opted for the approach above.

Cheers,

Joe

[Joseph Madden]

Hi all,

In the interest of explaining myself further and also considering a server scenario, i think it would be useful to be able to preserve the address information extracted from a DatagramPacket, where in a server application, it would be possible to use the address information from a request to send a response to a client.  I'd like to use the AddressedEnvelope to avoid polluting all message types with the address information.  To avoid having to write multiple handlers that would handle an AddressedEnvelope type for each of my messages, i think a generic message to message handler which handles AddressedEnvelopes and delegates to a provided message to message encoder/decoders would be useful.  Here is a snippet of what i mean ( "inspired" by DatagramPacketEncoder/Decoder)

public class AddressedEnvelopeMessageToMessageEncoder<M, T>

        extends MessageToMessageEncoder<AddressedEnvelope<M, InetSocketAddress>> {

    private final MessageToMessageEncoder<? super M> encoder;

    private final Class<T> resultClazz;

    /**

     * Create an encoder that encodes the content in {@link AddressedEnvelope}

     * to {@link AddressedEnvelope} using the specified message encoder.

     *

     * @param encoder

     *            the specified message encoder

     */

    public AddressedEnvelopeMessageToMessageEncoder(MessageToMessageEncoder<? super M> encoder,

            Class<T> resultClazz) {

        this.encoder = checkNotNull(encoder, "encoder");

        this.resultClazz = checkNotNull(resultClazz, "resultClazz");

    }

    @Override

    public boolean acceptOutboundMessage(Object msg) throws Exception {

        if (super.acceptOutboundMessage(msg)) {

            @SuppressWarnings("rawtypes")

            AddressedEnvelope envelope = (AddressedEnvelope) msg;

            return encoder.acceptOutboundMessage(envelope.content()) && envelope.sender() instanceof InetSocketAddress

                    && envelope.recipient() instanceof InetSocketAddress;

        }

        return false;

    }

    @Override

    protected void encode(ChannelHandlerContext ctx, AddressedEnvelope<M, InetSocketAddress> msg, List<Object> out)

            throws Exception {

        assert out.isEmpty();

        encoder.encode(ctx, msg.content(), out);

        if (out.size() != 1) {

            throw new EncoderException(StringUtil.simpleClassName(encoder) + " must produce only one message.");

        }

        Object content = out.get(0);

        if (resultClazz.isInstance(content)) {

            // Replace the T with an AddressedEnvelope<T, InetSocketAddress>.

            out.set(0, new DefaultAddressedEnvelope<T, InetSocketAddress>(resultClazz.cast(content), msg.recipient(),

                    msg.sender()));

        } else {

            throw new EncoderException(

                    StringUtil.simpleClassName(encoder) + " must produce only " + resultClazz.getSimpleName() + ".");

        }

    }

// ... methods removed for brevity

}

public class AddressedEnvelopeMessageToMessageDecoder<M, T>

        extends MessageToMessageDecoder<AddressedEnvelope<M, InetSocketAddress>> {

    private final MessageToMessageDecoder<M> decoder;

    private final Class<T> resultClazz;

    /**

     * Create a {@link DatagramPacket} decoder using the specified

     * {@link ByteBuf} decoder.

     *

     * @param decoder

     *            the specified {@link ByteBuf} decoder

     */

    public AddressedEnvelopeMessageToMessageDecoder(MessageToMessageDecoder<M> decoder, Class<T> resultClazz) {

        this.decoder = checkNotNull(decoder, "decoder");

        this.resultClazz = checkNotNull(resultClazz, "resultClazz");

    }

    @Override

    public boolean acceptInboundMessage(Object msg) throws Exception {

        if (msg instanceof AddressedEnvelope) {

            @SuppressWarnings("unchecked")

            AddressedEnvelope<M, InetSocketAddress> addressedEnvelope = (AddressedEnvelope<M, InetSocketAddress>) msg;

            return decoder.acceptInboundMessage(addressedEnvelope.content());

        }

        return false;

    }

    @Override

    protected void decode(ChannelHandlerContext ctx, AddressedEnvelope<M, InetSocketAddress> msg, List<Object> out)

            throws Exception {

        decoder.decode(ctx, msg.content(), out);

        if (out.size() != 1) {

            throw new EncoderException(StringUtil.simpleClassName(decoder) + " must produce only one message.");

        }

        Object content = out.get(0);

        if (resultClazz.isInstance(content)) {

            // Replace the T with an AddressedEnvelope<T, InetSocketAddress>.

            out.set(0, new DefaultAddressedEnvelope<T, InetSocketAddress>(resultClazz.cast(content), msg.recipient(),

                    msg.sender()));

        } else {

            throw new EncoderException(

                    StringUtil.simpleClassName(decoder) + " must produce only " + resultClazz.getSimpleName() + ".");

        }

    }

// .. methods removed for brevity

}

Regards,

Joe