senses

Author: ASKozienko

SnsClient.php

@@ -48,6 +48,16 @@ public function subscribe(array $args): Result
         return $this->callApi('subscribe', $args);
     }
 
+    public function unsubscribe(array $args): Result
+    {
+        return $this->callApi('unsubscribe', $args);
+    }
+
+    public function listSubscriptionsByTopic(array $args): Result
+    {
+        return $this->callApi('ListSubscriptionsByTopic', $args);
+    }
+
     public function getAWSClient(): AwsSnsClient
     {
         $this->resolveClient();

SnsContext.php

@@ -79,16 +79,61 @@ public function declareTopic(SnsDestination $destination): void
 
     public function subscribe(SnsSubscribe $subscribe): void
     {
+        foreach ($this->getSubscriptions($subscribe->getTopic()) as $subscription) {
+            if ($subscription['Protocol'] === $subscribe->getProtocol()
+                && $subscription['Endpoint'] === $subscribe->getEndpoint()) {
+                return;
+            }
+        }
+
         $this->client->subscribe([
             'Attributes' => $subscribe->getAttributes(),
             'Endpoint' => $subscribe->getEndpoint(),
-            //'Protocol' => '<string>', // REQUIRED
-//'ReturnSubscriptionArn' => true || false,
-//'TopicArn' => '<string>', // REQUIRED
-//]);
+            'Protocol' => $subscribe->getProtocol(),
+            'ReturnSubscriptionArn' => $subscribe->isReturnSubscriptionArn(),
+            'TopicArn' => $this->getTopicArn($subscribe->getTopic()),
         ]);
     }
 
+    public function unsubscibe(SnsUnsubscribe $unsubscribe): void
+    {
+        foreach ($this->getSubscriptions($unsubscribe->getTopic()) as $subscription) {
+            if ($subscription['Protocol'] != $unsubscribe->getProtocol()) {
+                continue;
+            }
+
+            if ($subscription['Endpoint'] != $unsubscribe->getEndpoint()) {
+                continue;
+            }
+
+            $this->client->unsubscribe([
+                'SubscriptionArn' => $subscription['SubscriptionArn'],
+            ]);
+        }
+    }
+
+    public function getSubscriptions(SnsDestination $destination): array
+    {
+        $args = [
+            'TopicArn' => $this->getTopicArn($destination),
+        ];
+
+        $subscriptions = [];
+        while (true) {
+            $result = $this->client->listSubscriptionsByTopic($args);
+
+            $subscriptions = array_merge($subscriptions, $result->get('Subscriptions'));
+
+            if (false == $result->hasKey('NextToken')) {
+                break;
+            }
+
+            $args['NextToken'] = $result->get('NextToken');
+        }
+
+        return $subscriptions;
+    }
+
     public function getTopicArn(SnsDestination $destination): string
     {
         if (false == array_key_exists($destination->getTopicName(), $this->topicArns)) {

SnsSubscribe.php

@@ -6,32 +6,67 @@
 
 class SnsSubscribe
 {
-    private $destination;
+    const PROTOCOL_SQS = 'sqs';
 
+    /**
+     * @var SnsDestination
+     */
+    private $topic;
+
+    /**
+     * @var string
+     */
     private $endpoint;
 
+    /**
+     * @var string
+     */
     private $protocol;
 
+    /**
+     * @var
+     */
+    private $returnSubscriptionArn;
+
+    /**
+     * @var
+     */
     private $attributes;
 
-    public function __construct(SnsDestination $destination, string $endpoint, string $protocol, array $arguments)
-    {
-        $this->destination = $destination;
+    public function __construct(
+        SnsDestination $topic,
+        string $endpoint,
+        string $protocol,
+        bool $returnSubscriptionArn = false,
+        array $attributes = []
+    ) {
+        $this->topic = $topic;
         $this->endpoint = $endpoint;
         $this->protocol = $protocol;
-        $this->attributes = $arguments;
+        $this->returnSubscriptionArn = $returnSubscriptionArn;
+        $this->attributes = $attributes;
     }
 
-    public function getDestination(): SnsDestination
+    public function getTopic(): SnsDestination
     {
-        return $this->destination;
+        return $this->topic;
     }
 
     public function getEndpoint(): string
     {
         return $this->endpoint;
     }
 
+    public function getProtocol(): string
+    {
+        return $this->protocol;
+    }
+
+    public function isReturnSubscriptionArn(): bool
+    {
+        return $this->returnSubscriptionArn;
+    }
+
     public function getAttributes(): array
     {
         return $this->attributes;

SnsUnsubscribe.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Enqueue\Sns;
+
+class SnsUnsubscribe
+{
+    /**
+     * @var SnsDestination
+     */
+    private $topic;
+
+    /**
+     * @var string
+     */
+    private $endpoint;
+
+    /**
+     * @var string
+     */
+    private $protocol;
+
+    public function __construct(
+        SnsDestination $topic,
+        string $endpoint,
+        string $protocol
+    ) {
+        $this->topic = $topic;
+        $this->endpoint = $endpoint;
+        $this->protocol = $protocol;
+    }
+
+    public function getTopic(): SnsDestination
+    {
+        return $this->topic;
+    }
+
+    public function getEndpoint(): string
+    {
+        return $this->endpoint;
+    }
+
+    public function getProtocol(): string
+    {
+        return $this->protocol;
+    }
+}