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

  /// Delay for which the spin loop should wait before starting over.
  let loopTimeout: TimeInterval

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

  /// Initializes a Server
  ///
  /// - Parameter address: the address where the server will listen
  /// - Parameter loopTimeout: delay for which the spin loop should wait before starting over.
  public init(address: String, loopTimeout: TimeInterval = 600) {
    underlyingServer = cgrpc_server_create(address)
    completionQueue = CompletionQueue(
      underlyingCompletionQueue: cgrpc_server_get_completion_queue(underlyingServer), name: "Server " + address)
    self.loopTimeout = loopTimeout
  }

  /// 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
  /// - Parameter rootCerts: used to validate client certificates; will enable enforcing valid client certificates when provided
  /// - Parameter loopTimeout: delay for which the spin loop should wait before starting over.
  public init(address: String, key: String, certs: String, rootCerts: String? = nil, loopTimeout: TimeInterval = 600) {
    underlyingServer = cgrpc_server_create_secure(address, key, certs, rootCerts, rootCerts == nil ? 0 : 1)
    completionQueue = CompletionQueue(
      underlyingCompletionQueue: cgrpc_server_get_completion_queue(underlyingServer), name: "Server " + address)
    self.loopTimeout = loopTimeout
  }

  deinit {
    cgrpc_server_destroy(underlyingServer)
    completionQueue.shutdown()
  }

  /// Run the server.
  ///
  /// - Parameter handlerFunction: will be called to handle an incoming request. Dispatched on a new thread, so can be blocking.
  public func run(handlerFunction: @escaping (Handler) -> Void) {
    cgrpc_server_start(underlyingServer)
    // run the server on a new background thread
    let spinloopThreadQueue = DispatchQueue(label: "SwiftGRPC.CompletionQueue.runToCompletion.spinloopThread")
    spinloopThreadQueue.async {
      do {
        // Allocate a handler _outside_ the spin loop, as we must use _this particular_ handler to serve the next call
        // once we have called `handler.requestCall`. In particular, we need to keep the current handler for the next
        // spin loop interation when we hit the `.queueTimeout` case. The handler should only be replaced once it is
        // "used up" for serving an incoming call.
        var handler = Handler(underlyingServer: self.underlyingServer)
        // Tell gRPC to store the next call's information in this handler object.
        try handler.requestCall(tag: Server.handlerCallTag)
        var spinloopActive = true
        while spinloopActive {
          try withAutoReleasePool {
            // block while waiting for an incoming request
            let event = self.completionQueue.wait(timeout: self.loopTimeout)

            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
                  var strongHandlerReference: Handler?
                  strongHandlerReference = handler
                  // To prevent the "Variable 'strongHandlerReference' was written to, but never read" warning.
                  _ = strongHandlerReference
                  // this will start the completion queue on a new thread
                  handler.completionQueue.runToCompletion {
                    // release the handler when it finishes
                    strongHandlerReference = nil
                  }

                  // Dispatch the handler function on a separate thread.
                  let handlerDispatchThreadQueue = DispatchQueue(label: "SwiftGRPC.Server.run.dispatchHandlerThread")
                  // Needs to be copied, because we will change the value of `handler` right after this.
                  let handlerCopy = handler
                  handlerDispatchThreadQueue.async {
                    handlerFunction(handlerCopy)
                  }
                }

                // This handler has now been "used up" for the current call; replace it with a fresh one for the next
                // loop iteration.
                handler = Handler(underlyingServer: self.underlyingServer)
                try handler.requestCall(tag: Server.handlerCallTag)
              } else if event.tag == Server.stopTag || event.tag == Server.destroyTag {
                spinloopActive = false
                return
              }
            } else if event.type == .queueTimeout {
              // Everything is fine, just start over *while continuing to use the existing handler*.
              return
            } else if event.type == .queueShutdown {
              spinloopActive = false
              return
            }
          }
        }
      } catch {
        print("server call error: \(error)")
      }
      self.onCompletion?()
    }
  }

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