Modified to use topics instead of queues.