This should probably be written as two modules, one for the producer and another for the consumer, since they use different types of connection. It's questionable whether PHP-Queue is a healthy layer. There's only a benefit if it's easy to swap out Kafka for another queue provider, so avoid creating any new custom APIs, or if we do, implement these functions for the other queues.