grpc-swift/Sources/SwiftGRPC/Core/Server.swift

/*
 * Copyright 2016, gRPC Authors All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#if SWIFT_PACKAGE
  import CgRPC
  import Dispatch
#endif
import Foundation

/// gRPC Server
public class Server {
  static let handlerCallTag = 101
  
  // These are sent by the CgRPC shim.
  static let stopTag = 0
  static let destroyTag = 1000
  
  /// Pointer to underlying C representation
  private let underlyingServer: UnsafeMutableRawPointer

  /// Completion queue used for server operations
  let completionQueue: CompletionQueue

  /// Active handlers
  private var handlers = Set<Handler>()

  /// Mutex for synchronizing access to handlers
  private let handlersMutex: Mutex = Mutex()

  /// Optional callback when server stops serving
  public var onCompletion: (() -> Void)?

  /// Initializes a Server
  ///
  /// - Parameter address: the address where the server will listen
  public init(address: String) {
    underlyingServer = cgrpc_server_create(address)
    completionQueue = CompletionQueue(
      underlyingCompletionQueue: cgrpc_server_get_completion_queue(underlyingServer), name: "Server " + address)
  }

  /// Initializes a secure Server
  ///
  /// - Parameter address: the address where the server will listen
  /// - Parameter key: the private key for the server's certificates
  /// - Parameter certs: the server's certificates
  public init(address: String, key: String, certs: String) {
    underlyingServer = cgrpc_server_create_secure(address, key, certs)
    completionQueue = CompletionQueue(
      underlyingCompletionQueue: cgrpc_server_get_completion_queue(underlyingServer), name: "Server " + address)
  }

  deinit {
    cgrpc_server_destroy(underlyingServer)
  }

  /// Run the server
  public func run(dispatchQueue: DispatchQueue = DispatchQueue.global(),
                  handlerFunction: @escaping (Handler) -> Void) {
    cgrpc_server_start(underlyingServer)
    // run the server on a new background thread
    dispatchQueue.async {
      var running = true
      while running {
        do {
          let handler = Handler(underlyingServer: self.underlyingServer)
          try handler.requestCall(tag: Server.handlerCallTag)

          // block while waiting for an incoming request
          let event = self.completionQueue.wait(timeout: 600)

          if event.type == .complete {
            if event.tag == Server.handlerCallTag {
              // run the handler and remove it when it finishes
              if event.success != 0 {
                // hold onto the handler while it runs
                self.handlersMutex.synchronize {
                  self.handlers.insert(handler)
                }
                // this will start the completion queue on a new thread
                handler.completionQueue.runToCompletion(callbackQueue: dispatchQueue) {
                  dispatchQueue.async {
                    self.handlersMutex.synchronize {
                      // release the handler when it finishes
                      self.handlers.remove(handler)
                    }
                  }
                }
                // call the handler function on the server thread
                handlerFunction(handler)
              }
            } else if event.tag == Server.stopTag || event.tag == Server.destroyTag {
              running = false // exit the loop
            }
          } else if event.type == .queueTimeout {
            // everything is fine
          } else if event.type == .queueShutdown {
            running = false
          }
        } catch {
          print("server call error: \(error)")
          running = false
        }
      }
      self.onCompletion?()
    }
  }

  public func stop() {
    cgrpc_server_stop(underlyingServer)
  }
}