From c1b74c39568d6152f1498508f5aaaaf069aba39b Mon Sep 17 00:00:00 2001
From: Zach Smith <zach@driver.xyz>
Date: Tue, 2 Oct 2018 16:53:12 -0700
Subject: Fixes for alicloud pubsub (#229)

* Remove recover from processMessage in StreatBus

* Check if queue exists before creating
---
 .../xyz/driver/core/messaging/AliyunBus.scala      | 30 ++++++++++++----------
 .../xyz/driver/core/messaging/StreamBus.scala      |  2 +-
 2 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala b/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
index c23ea0f..8b7bca7 100644
--- a/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
+++ b/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
@@ -136,18 +136,22 @@ class AliyunBus(
 
   def createSubscription(topic: Topic[_], config: SubscriptionConfig): Future[Unit] = Future {
     val subscriptionName = rawSubscriptionName(config, topic)
-    val topicName        = rawTopicName(topic)
-    val topicRef         = client.getTopicRef(topicName)
-
-    val queueMeta = new QueueMeta
-    queueMeta.setQueueName(subscriptionName)
-    queueMeta.setVisibilityTimeout(config.ackTimeout.toSeconds)
-    client.createQueue(queueMeta)
-
-    val subscriptionMeta = new SubscriptionMeta
-    subscriptionMeta.setSubscriptionName(subscriptionName)
-    subscriptionMeta.setTopicName(topicName)
-    subscriptionMeta.setEndpoint(topicRef.generateQueueEndpoint(subscriptionName))
-    topicRef.subscribe(subscriptionMeta)
+    val queueExists      = Option(client.listQueue(subscriptionName, "", 1)).exists(!_.getResult.isEmpty)
+
+    if (!queueExists) {
+      val topicName = rawTopicName(topic)
+      val topicRef  = client.getTopicRef(topicName)
+
+      val queueMeta = new QueueMeta
+      queueMeta.setQueueName(subscriptionName)
+      queueMeta.setVisibilityTimeout(config.ackTimeout.toSeconds)
+      client.createQueue(queueMeta)
+
+      val subscriptionMeta = new SubscriptionMeta
+      subscriptionMeta.setSubscriptionName(subscriptionName)
+      subscriptionMeta.setTopicName(topicName)
+      subscriptionMeta.setEndpoint(topicRef.generateQueueEndpoint(subscriptionName))
+      topicRef.subscribe(subscriptionMeta)
+    }
   }
 }
diff --git a/src/main/scala/xyz/driver/core/messaging/StreamBus.scala b/src/main/scala/xyz/driver/core/messaging/StreamBus.scala
index a9ba3a7..44d75cd 100644
--- a/src/main/scala/xyz/driver/core/messaging/StreamBus.scala
+++ b/src/main/scala/xyz/driver/core/messaging/StreamBus.scala
@@ -76,7 +76,7 @@ trait StreamBus extends Bus {
         maxRestarts
       ) { () =>
         subscribe(topic, config)
-          .via(processMessage.recover({ case _ => Nil }))
+          .via(processMessage)
           .log(topic.name)
           .mapConcat(identity)
       }
-- 
cgit v1.2.3