Merge pull request #219 from MrMage/synchronize-stubs

Make the test stubs thread safe.

Sources/SwiftGRPC/Runtime/ClientCallBidirectionalStreaming.swift

@@ -54,14 +54,18 @@ open class ClientCallBidirectionalStreamingBase<InputType: Message, OutputType:
 open class ClientCallBidirectionalStreamingTestStub<InputType: Message, OutputType: Message>: ClientCallBidirectionalStreaming {
   open class var method: String { fatalError("needs to be overridden") }
 
+  open var lock = Mutex()
+  
   open var inputs: [InputType] = []
   open var outputs: [OutputType] = []
   
   public init() {}
 
   open func _receive(timeout: DispatchTime) throws -> OutputType? {
-    defer { if !outputs.isEmpty { outputs.removeFirst() } }
-    return outputs.first
+    return lock.synchronize {
+      defer { if !outputs.isEmpty { outputs.removeFirst() } }
+      return outputs.first
+    }
   }
   
   open func receive(completion: @escaping (ResultOrRPCError<OutputType?>) -> Void) throws {
@@ -69,11 +73,11 @@ open class ClientCallBidirectionalStreamingTestStub<InputType: Message, OutputTy
   }
 
   open func send(_ message: InputType, completion _: @escaping (Error?) -> Void) throws {
-    inputs.append(message)
+    lock.synchronize { inputs.append(message) }
   }
   
   open func _send(_ message: InputType, timeout: DispatchTime) throws {
-    inputs.append(message)
+    lock.synchronize { inputs.append(message) }
   }
 
   open func closeSend(completion: (() -> Void)?) throws { completion?() }

Sources/SwiftGRPC/Runtime/ClientCallClientStreaming.swift

@@ -67,17 +67,19 @@ open class ClientCallClientStreamingBase<InputType: Message, OutputType: Message
 open class ClientCallClientStreamingTestStub<InputType: Message, OutputType: Message>: ClientCallClientStreaming {
   open class var method: String { fatalError("needs to be overridden") }
 
+  open var lock = Mutex()
+  
   open var inputs: [InputType] = []
   open var output: OutputType?
   
   public init() {}
 
   open func send(_ message: InputType, completion _: @escaping (Error?) -> Void) throws {
-    inputs.append(message)
+    lock.synchronize { inputs.append(message) }
   }
   
   open func _send(_ message: InputType, timeout: DispatchTime) throws {
-    inputs.append(message)
+    lock.synchronize { inputs.append(message) }
   }
 
   open func closeAndReceive(completion: @escaping (ResultOrRPCError<OutputType>) -> Void) throws {

Sources/SwiftGRPC/Runtime/ClientCallServerStreaming.swift

@@ -41,13 +41,17 @@ open class ClientCallServerStreamingBase<InputType: Message, OutputType: Message
 open class ClientCallServerStreamingTestStub<OutputType: Message>: ClientCallServerStreaming {
   open class var method: String { fatalError("needs to be overridden") }
 
+  open var lock = Mutex()
+  
   open var outputs: [OutputType] = []
   
   public init() {}
   
   open func _receive(timeout: DispatchTime) throws -> OutputType? {
-    defer { if !outputs.isEmpty { outputs.removeFirst() } }
-    return outputs.first
+    return lock.synchronize {
+      defer { if !outputs.isEmpty { outputs.removeFirst() } }
+      return outputs.first
+    }
   }
   
   open func receive(completion: @escaping (ResultOrRPCError<OutputType?>) -> Void) throws {

Sources/SwiftGRPC/Runtime/ServerSessionBidirectionalStreaming.swift

@@ -65,13 +65,17 @@ open class ServerSessionBidirectionalStreamingBase<InputType: Message, OutputTyp
 /// Simple fake implementation of ServerSessionBidirectionalStreaming that returns a previously-defined set of results
 /// and stores sent values for later verification.
 open class ServerSessionBidirectionalStreamingTestStub<InputType: Message, OutputType: Message>: ServerSessionTestStub, ServerSessionBidirectionalStreaming {
+  open var lock = Mutex()
+  
   open var inputs: [InputType] = []
   open var outputs: [OutputType] = []
   open var status: ServerStatus?
 
   open func _receive(timeout: DispatchTime) throws -> InputType? {
-    defer { if !inputs.isEmpty { inputs.removeFirst() } }
-    return inputs.first
+    return lock.synchronize {
+      defer { if !inputs.isEmpty { inputs.removeFirst() } }
+      return inputs.first
+    }
   }
   
   open func receive(completion: @escaping (ResultOrRPCError<InputType?>) -> Void) throws {
@@ -79,15 +83,15 @@ open class ServerSessionBidirectionalStreamingTestStub<InputType: Message, Outpu
   }
 
   open func send(_ message: OutputType, completion _: @escaping (Error?) -> Void) throws {
-    outputs.append(message)
+    lock.synchronize { outputs.append(message) }
   }
 
   open func _send(_ message: OutputType, timeout: DispatchTime) throws {
-    outputs.append(message)
+    lock.synchronize { outputs.append(message) }
   }
 
   open func close(withStatus status: ServerStatus, completion: (() -> Void)?) throws {
-    self.status = status
+    lock.synchronize { self.status = status }
     completion?()
   }
 

Sources/SwiftGRPC/Runtime/ServerSessionClientStreaming.swift

@@ -71,13 +71,17 @@ open class ServerSessionClientStreamingBase<InputType: Message, OutputType: Mess
 /// Simple fake implementation of ServerSessionClientStreaming that returns a previously-defined result
 /// and stores sent values for later verification.
 open class ServerSessionClientStreamingTestStub<InputType: Message, OutputType: Message>: ServerSessionTestStub, ServerSessionClientStreaming {
+  open var lock = Mutex()
+  
   open var inputs: [InputType] = []
   open var output: OutputType?
   open var status: ServerStatus?
 
   open func _receive(timeout: DispatchTime) throws -> InputType? {
-    defer { if !inputs.isEmpty { inputs.removeFirst() } }
-    return inputs.first
+    return lock.synchronize {
+      defer { if !inputs.isEmpty { inputs.removeFirst() } }
+      return inputs.first
+    }
   }
   
   open func receive(completion: @escaping (ResultOrRPCError<InputType?>) -> Void) throws {
@@ -85,13 +89,17 @@ open class ServerSessionClientStreamingTestStub<InputType: Message, OutputType:
   }
 
   open func sendAndClose(response: OutputType, status: ServerStatus, completion: (() -> Void)?) throws {
-    self.output = response
-    self.status = status
+    lock.synchronize {
+      self.output = response
+      self.status = status
+    }
     completion?()
   }
 
   open func sendErrorAndClose(status: ServerStatus, completion: (() -> Void)? = nil) throws {
-    self.status = status
+    lock.synchronize {
+      self.status = status
+    }
     completion?()
   }
   

Sources/SwiftGRPC/Runtime/ServerSessionServerStreaming.swift

@@ -65,19 +65,21 @@ open class ServerSessionServerStreamingBase<InputType: Message, OutputType: Mess
 /// Simple fake implementation of ServerSessionServerStreaming that returns a previously-defined set of results
 /// and stores sent values for later verification.
 open class ServerSessionServerStreamingTestStub<OutputType: Message>: ServerSessionTestStub, ServerSessionServerStreaming {
+  open var lock = Mutex()
+  
   open var outputs: [OutputType] = []
   open var status: ServerStatus?
 
   open func send(_ message: OutputType, completion _: @escaping (Error?) -> Void) throws {
-    outputs.append(message)
+    lock.synchronize { outputs.append(message) }
   }
 
   open func _send(_ message: OutputType, timeout: DispatchTime) throws {
-    outputs.append(message)
+    lock.synchronize { outputs.append(message) }
   }
 
   open func close(withStatus status: ServerStatus, completion: (() -> Void)?) throws {
-    self.status = status
+    lock.synchronize { self.status = status }
     completion?()
   }