From 3eb6a9e96bd8bf111490f390ea94a1c6d7677eff Mon Sep 17 00:00:00 2001
From: Zach Smith <zach@driver.xyz>
Date: Wed, 12 Sep 2018 13:34:57 -0700
Subject: Mask InternalServerError and MessageNotExist errors in AliyunBus
 (#218)

---
 .../xyz/driver/core/messaging/AliyunBus.scala      | 23 +++++++++++++++-------
 1 file changed, 16 insertions(+), 7 deletions(-)

diff --git a/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala b/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
index 92e47bd..c23ea0f 100644
--- a/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
+++ b/src/main/scala/xyz/driver/core/messaging/AliyunBus.scala
@@ -16,11 +16,17 @@ class AliyunBus(
     accessSecret: String,
     region: String,
     namespace: String,
-    pullTimeout: Int)(implicit val executionContext: ExecutionContext)
+    pullTimeout: Int
+)(implicit val executionContext: ExecutionContext)
     extends Bus {
-  val endpoint     = s"https://$accountId.mns.$region.aliyuncs.com"
-  val cloudAccount = new CloudAccount(accessId, accessSecret, endpoint)
-  val client       = cloudAccount.getMNSClient
+  private val endpoint     = s"https://$accountId.mns.$region.aliyuncs.com"
+  private val cloudAccount = new CloudAccount(accessId, accessSecret, endpoint)
+  private val client       = cloudAccount.getMNSClient
+
+  // When calling the asyncBatchPopMessage endpoint, alicloud returns an error if no message is received before the
+  // pullTimeout. This error is documented as MessageNotExist, however it's been observed to return InternalServerError
+  // occasionally. We mask both of these errors and return an empty list of messages
+  private val MaskedErrorCodes: Set[String] = Set("MessageNotExist", "InternalServerError")
 
   override val defaultMaxMessages: Int = 10
 
@@ -53,11 +59,14 @@ class AliyunBus(
       maxMessages,
       pullTimeout,
       new AsyncCallback[util.List[model.Message]] {
-        override def onSuccess(result: util.List[model.Message]): Unit = promise.success(result.asScala)
+        override def onSuccess(result: util.List[model.Message]): Unit = {
+          promise.success(result.asScala)
+        }
         override def onFail(ex: Exception): Unit = ex match {
-          case serviceException: ServiceException if serviceException.getErrorCode == "MessageNotExist" =>
+          case serviceException: ServiceException if MaskedErrorCodes(serviceException.getErrorCode) =>
             promise.success(Nil)
-          case _ => promise.failure(ex)
+          case _ =>
+            promise.failure(ex)
         }
       }
     )
-- 
cgit v1.2.3