WIP: add message consumer

kaliop-uk · Sep 1, 2015 · 8e85df5 · 8e85df5
1 parent 1d20f58
commit 8e85df5
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 5 deletions.
diff --git a/Adapter/Kinesis/Consumer.php b/Adapter/Kinesis/Consumer.php
@@ -0,0 +1,35 @@
+<?php
+
+namespace Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis;
+
+
+class Consumer
+{
+    protected $shard;
+    protected $streamName;
+
+    /**
+     * Does nothing
+     */
+    public function setMemoryLimit()
+    {
+    }
+
+    public function setRoutingKey($key)
+    {
+        $this->shard = $key;
+    }
+
+    public function consume($amount)
+    {
+// @todo
+    }
+
+    /**
+     * @param string $queueName
+     */
+    public function setQueueName($queueName)
+    {
+        $this->streamName = $queueName;
+    }
+}
diff --git a/Adapter/Kinesis/Driver.php b/Adapter/Kinesis/Driver.php
@@ -9,6 +9,13 @@ class Driver extends ContainerAware implements DriverInterface
 {
     protected $debug;
 
+    public function getConsumer($queueName)
+    {
+        $consumer = $this->container->get('kaliop_queueing.kinesis.message_consumer');
+        $consumer->setQueueName($queueName);
+        return $consumer;
+    }
+
     public function acceptMessage($message)
     {
 // @todo
@@ -25,9 +32,9 @@ public function decodeMessage($message)
 
     /**
      * @param string $queueName
-     * @return \Kaliop\QueueingBundle\Queue\MessageProducerInterface
+     * @return \Kaliop\QueueingBundle\Queue\ProducerInterface
      */
-    public function getMessageProducer($queueName)
+    public function getProducer($queueName)
     {
         $producer = $this->container->get('kaliop_queueing.kinesis.message_producer');
         $producer->setQueueName($queueName);

diff --git a/Adapter/Kinesis/Producer.php b/Adapter/Kinesis/Producer.php
@@ -2,10 +2,10 @@
 
 namespace Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis;
 
-use Kaliop\QueueingBundle\Queue\MessageProducerInterface;
+use Kaliop\QueueingBundle\Queue\ProducerInterface;
 use Aws\Kinesis\KinesisClient;
 
-class Producer implements MessageProducerInterface
+class Producer implements ProducerInterface
 {
     /** @var  \Aws\Kinesis\KinesisClient */
     protected $client;

diff --git a/Resources/config/services.yml b/Resources/config/services.yml
@@ -2,6 +2,7 @@ parameters:
     kaliop_queueing.driver.kinesis.class: Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Driver
     kaliop_queueing.kinesis.queue_manager.class: Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\QueueManager
     kaliop_queueing.kinesis.message_producer.class: Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Producer
+    kaliop_queueing.kinesis.message_consumer.class: Kaliop\Queueing\Plugins\KinesisBundle\Adapter\Kinesis\Consumer
 
 services:
     kaliop_queueing.driver.kinesis:
@@ -19,4 +20,8 @@ services:
 
     kaliop_queueing.kinesis.message_producer:
         class: %kaliop_queueing.kinesis.message_producer.class%
-        arguments: [ %kaliop_queueing_kinesis.default.client_config% ]
+        arguments: [ %kaliop_queueing_kinesis.default.client_config% ]
+
+    kaliop_queueing.kinesis.message_consumer:
+        class: %kaliop_queueing.kinesis.message_consumer.class%
+        #arguments: [ %kaliop_queueing_kinesis.default.client_config% ]