Fix heartbeats on Android when CPU sleeps

[youngj]

We're using RabbitMQ along with the Java client library for http://telerivet.com/ to send push messages to Android phones.

For the most part, RabbitMQ client library works great for push messaging on Android, except that heartbeats don't work when the CPU sleeps. It's important to let the CPU sleep as much as possible to maximize battery life -- for example the phone would run for only 10 hours if we use a WakeLock to keep the CPU on, but it could last for 3 days if we let the CPU sleep most of the time and just wake it up every 5 minutes to send a heartbeat.

There are a few assumptions in com.rabbitmq.client.impl.HeartbeatSender that break on Android when the CPU sleeps:

1. Executors.newSingleThreadScheduledExecutor doesn't wake up Android if the CPU is asleep
2. The optimization to skip sending heartbeats after recent activity doesn't work on Android because System.nanoTime() doesn't include the time that the CPU is asleep.
3. The optimization to skip sending heartbeats after recent activity is actually less efficient on Android because it requires waking the CPU 2x as often.

To fix the first issue, I added a method ConnectionFactory.setHeartbeatExecutor that allows application code on Android to pass in an executor that wraps Android's AlarmManager to wake up the CPU if it's asleep.

Here are the relevant parts from my application code as an example:

public class AmqpConsumer {
    ...
    private Runnable heartbeatRunnable;
    private AlarmManager alarmManager;
    ...

    public synchronized startBlocking()
    {
        ...
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHeartbeatExecutor(new HeartbeatExecutor());
        ...
    }

    public synchronized void setHeartbeatRunnable(Runnable heartbeatRunnable)
    {
        this.heartbeatRunnable = heartbeatRunnable;
    }

    public synchronized void sendHeartbeatBlocking()
    {
        if (heartbeatRunnable != null)
        {
            heartbeatRunnable.run();
        }        
    }

    public PendingIntent getHeartbeatPendingIntent()
    {
        return PendingIntent.getService(app, 0, 
             new Intent(app, AmqpHeartbeatService.class), 0);
    }                    

    public class HeartbeatExecutor extends ScheduledThreadPoolExecutor
    {
        public HeartbeatExecutor()
        {
            super(1);
        }        

        @Override
        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, 
                long initialDelay, long period, TimeUnit unit) 
        {            
            long delayMs = TimeUnit.MILLISECONDS.convert(initialDelay, unit);
            long periodMs = TimeUnit.MILLISECONDS.convert(period, unit);

            setHeartbeatRunnable(command);

            alarmManager.setRepeating(
                AlarmManager.ELAPSED_REALTIME_WAKEUP,
                SystemClock.elapsedRealtime() + delayMs,
                periodMs,
                getHeartbeatPendingIntent());

            return new ScheduledFuture<Integer>()
            {
                private boolean cancelled;

                public long getDelay(TimeUnit u2) { return 0; }
                public int compareTo(Delayed d2) { return 0; }
                public Integer get() { return null; }
                public Integer get(long timeout, TimeUnit unit) { return null; }
                public boolean cancel(boolean interrupt)
                {
                    alarmManager.cancel(getHeartbeatPendingIntent());
                    cancelled = true;
                    return true;
                }            

                public boolean isCancelled() { return cancelled; }
                public boolean isDone() { return cancelled; }
            };
        }
    }
}

public class AmqpHeartbeatService extends IntentService {
    ...        
    @Override
    protected void onHandleIntent(Intent intent)
    {          
        App app = (App)this.getApplicationContext();        
        AmqpConsumer consumer = app.getAmqpConsumer();
        consumer.sendHeartbeatBlocking();
    }
}

The application code to implement ScheduledExecutorService is pretty tedious, but it only requires a small change in the RabbitMQ Java client library to use a custom executor instead of Executors.newSingleThreadScheduledExecutor().

Also, I removed the optimization to skip sending heartbeats if there has been recent activity. It doesn't work on Android because System.nanoTime() doesn't include the time that the CPU is asleep. As a result, the client will miss heartbeats and the server will shut down the connection.

On Android, real time can be determined with SystemClock.elapsedRealtime(), but this isn't cross-platform.

Also, the optimization to skip sending heartbeats after recent activity was actually less efficient on Android because it requires waking the CPU up 2x as often. To maximize battery life, it is better to wake up the CPU the absolute minimum necessary (even if it may require sending more heartbeat frames if there has been recent activity).

[rade]

Apologies for taking to so long to respond to this.

I'm happy with the changes to make the heartbeat executor configurable, but wouldn't want to lose the "skip sending heartbeats when there has been recent activity" optimisation. So we either need to find a way to make that work on Android (and other systems that put the CPU to sleep frequently), or provide a way of disabling it.

[michaelklishin]

Some of these changes are not necessarily a great fit for every user and this PR no longer merges clearly. Closing as too old. Let us know if this is still relevant and if there are better ways to work around it.