Test update: expand and update stop sending after an error is received.

grpc · Apr 8, 2019 · d99e82a · d99e82a
1 parent 7a69134
commit d99e82a
Showing 1 changed file with 34 additions and 4 deletions.
diff --git a/Sources/Examples/Echo/EchoProvider.swift b/Sources/Examples/Echo/EchoProvider.swift
@@ -27,15 +27,20 @@ class EchoProvider: Echo_EchoProvider {
 
   // expand splits a request into words and returns each word in a separate message.
   func expand(request: Echo_EchoRequest, session: Echo_EchoExpandSession) throws -> ServerStatus? {
+    let sendError = Atomic<Bool>(false)
     let parts = request.text.components(separatedBy: " ")
     for (i, part) in parts.enumerated() {
       var response = Echo_EchoResponse()
       response.text = "Swift echo expand (\(i)): \(part)"
       try session.send(response) {
         if let error = $0 {
-          print("expand error: \(error)")
+          print("expand send error: \(error)")
+          sendError.mutate { $0 = true }
         }
       }
+      if sendError.value {
+        break
+      }
     }
     return .ok
   }
@@ -49,7 +54,7 @@ class EchoProvider: Echo_EchoProvider {
           else { break }  // End of stream
         parts.append(request.text)
       } catch {
-        print("collect error: \(error)")
+        print("collect receive error: \(error)")
         break
       }
     }
@@ -60,6 +65,7 @@ class EchoProvider: Echo_EchoProvider {
 
   // update streams back messages as they are received in an input stream.
   func update(session: Echo_EchoUpdateSession) throws -> ServerStatus? {
+    let sendError = Atomic<Bool>(false)
     var count = 0
     while true {
       do {
@@ -70,14 +76,38 @@ class EchoProvider: Echo_EchoProvider {
         count += 1
         try session.send(response) {
           if let error = $0 {
-            print("update error: \(error)")
+            print("update send error: \(error)")
+            sendError.mutate { $0 = true }
           }
         }
+        if sendError.value {
+          break
+        }
       } catch {
-        print("update error: \(error)")
+        print("update receive error: \(error)")
         break
       }
     }
     return .ok
   }
 }
+
+final class Atomic<A> {
+  private let queue = DispatchQueue(label: "Atomic serial queue")
+  private var _value: A
+  init(_ value: A) {
+    self._value = value
+  }
+
+  var value: A {
+    get {
+      return queue.sync { self._value }
+    }
+  }
+
+  func mutate(_ transform: (inout A) -> ()) {
+    queue.sync {
+      transform(&self._value)
+    }
+  }
+}