RabbbitMQ RpcServer and RpcClient recovery fail about lyra

[jhalterman]

I've never actually used the RpcClient so I'm not surprised that it's not supported. Just looking at the code real quick though, it looks like an internal reference to the consumer is wiped when the channel shutdown occurs. You may need to configure ModelMapper to not recover the consumer:

confit.withConsumerRecovery(false);

...instead letting the Rpc client re-create the consumer after the channel is recovered:

config.withChannelListeners(new AbstractChannelListener() {
    public void onRecovery(Channel channel) {
        rpcClient.setupConsumer();
    }
});

I haven't tested this myself (busy on other things ATM), but let me know how it goes.

[chaopeng]

My fail... lyra can recovery rpc automatically. The exception because rabbitmq restart...

[jhalterman]

Sure, but even after a RabbitMQ restart hopefully you can reuse your RpcClient. Did that approach I pasted above work?

[chaopeng]

if it is restart, should re-create server and client on onRecovery() ....

[jhalterman]

On recovery will be called after the connection and channel have been recovered. This is where you'll need to tell the client to recover the consumer since RpcClient needs that to be done internally.

[chaopeng]

it works now, thank you for your help.

[jhalterman]

Did you end up using the channel listener approach?

[chaopeng]

public class RecoveryRabbitMQRpcServer {

    private RpcServer rpcServer;
    private ExecutorService es = Executors.newSingleThreadExecutor();

    public RecoveryRabbitMQRpcServer(String host, String vhost, String userName, String password, final String queue, final Class<? extends RpcServer> serverCls) throws IOException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
        Config config = new Config()
                .withRecoveryPolicy(new RecoveryPolicy()
                        .withMaxAttempts(20)
                        .withInterval(Duration.seconds(1))
                        .withMaxDuration(Duration.minutes(5)));

        config.withConsumerRecovery(false);
        config.withChannelListeners(new DefaultChannelListener() {
            @Override
            public void onRecovery(Channel channel) {
                try {
                    rpcServer = serverCls.getConstructor(new Class[]{Channel.class, String.class}).newInstance(channel, queue);
                    es.submit(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                rpcServer.mainloop();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        ConnectionOptions options = new ConnectionOptions().withHost(host).withVirtualHost(vhost).withUsername(userName).withPassword(password);
        ConfigurableConnection connection = Connections.create(options, config);

        Channel channel = connection.createChannel();

        channel.queueDeclare(queue, false, false, false, null);

        this.rpcServer = serverCls.getConstructor(new Class[]{Channel.class, String.class}).newInstance(channel, queue);
        es.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    rpcServer.mainloop();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });

    }

}

public class RecoveryRabbitMQRpcClient {

    private Channel channel;
    private RpcClient rpcClient;

    /**
     *
     * @param timeout -1==NO_TIMEOUT
     */
    public RecoveryRabbitMQRpcClient(String host, String vhost, String userName, String password, final String queue, final int timeout) throws IOException {
        Config config = new Config()
                .withRecoveryPolicy(new RecoveryPolicy()
                        .withMaxAttempts(20)
                        .withInterval(Duration.seconds(1))
                        .withMaxDuration(Duration.minutes(5)));

        config.withConsumerRecovery(false);
        config.withChannelListeners(new DefaultChannelListener() {
            @Override
            public void onRecovery(Channel channel) {
                super.onRecovery(channel);
                try {
                    rpcClient = new RpcClient(channel, "", queue, timeout);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });

        ConnectionOptions options = new ConnectionOptions().withHost(host).withVirtualHost(vhost).withUsername(userName).withPassword(password);
        ConfigurableConnection connection = Connections.create(options, config);

        this.channel = connection.createChannel();


        this.rpcClient = new RpcClient(channel, "", queue, timeout);
    }

    public String stringCall(String s) throws IOException, TimeoutException {
        return this.rpcClient.stringCall(s);
    }

}

[jhalterman]

Thanks. I actually compiled the approach I was trying to get at and posted it here:

https://github.com/jhalterman/lyra/wiki/Lyra-Cookbook#recoverable-rpcclient