final int count = 4;
final long sleep = 500;
final Integer id = new Integer(1);
final ObjectName name = ObjectName.getInstance(":name=emitter");
final MutableBoolean notify = new MutableBoolean(true);
final MutableLong queued = new MutableLong(0);
final MutableLong delivered = new MutableLong(0);
Map environment = new HashMap();
environment.put(MX4JRemoteConstants.NOTIFICATION_QUEUE_CAPACITY, new Integer(queueCapacity));
RemoteNotificationClientHandler handler = new AbstractRemoteNotificationClientHandler(null, null, environment)
{
protected NotificationResult fetchNotifications(long sequenceNumber, int maxNumber, long timeout)
{
if (sequenceNumber < 0) return new NotificationResult(0, 0, new TargetedNotification[0]);
boolean doNotify = false;
synchronized (lock)
{
doNotify = notify.get();
}
if (doNotify)
{
// Avoid spin looping the fetcher thread, but don't sleep too much, we have to fill the client's queue
sleep(sleep);
TargetedNotification[] notifications = new TargetedNotification[count];
for (int i = 0; i < count; ++i) notifications[i] = new TargetedNotification(new Notification("type", name, sequenceNumber + i), id);
long nextSequence = sequenceNumber + count;
NotificationResult result = new NotificationResult(0, nextSequence, notifications);
synchronized (lock)
{
queued.set(getNotificationsCount());
}
return result;
}
else
{
sleep(timeout);
return new NotificationResult(0, sequenceNumber, new TargetedNotification[0]);
}
}
protected long getRetryPeriod()
{
return 1000;
}
protected int getMaxRetries()
{
return 5;
}
protected void sendConnectionNotificationLost(long number)
{
System.out.println("Lost notifications: " + number);
// Stop sending notifications
synchronized (lock)
{
notify.set(false);
// Deliver notifications until the last we queued on the client
queued.set(getNotificationsCount());
}
}
};
NotificationListener listener = new NotificationListener()
{
public void handleNotification(Notification notification, Object handback)
{
long sequence = notification.getSequenceNumber();
synchronized (lock)
{
delivered.set(sequence);
}
System.out.println("Received notification, sequence is " + sequence);
// Sleep longer than notification emission, to fill the client's queue
sleep(sleep * 2);
System.out.println("Handled notification, sequence is " + sequence);
}
};
try
{
handler.start();
handler.addNotificationListener(id, new NotificationTuple(name, listener, null, null));
// Wait until we empty the client's queue
synchronized (lock)
{
while (notify.get())
{
lock.wait(50);
if (queued.get() > queueCapacity) fail("Queued notifications " + queued.get() + " must not pass max capacity " + queueCapacity);
}
// Test timeouts if we don't deliver everything
while (delivered.get() < queued.get()) lock.wait(10);
}
}
finally
{
handler.stop();