1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
|
//===----------------------------------------------------------------------===//
//
// This source file is part of the SwiftNIO open source project
//
// Copyright (c) 2021 Apple Inc. and the SwiftNIO project authors
// Licensed under Apache License v2.0
//
// See LICENSE.txt for license information
// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
//
// SPDX-License-Identifier: Apache-2.0
//
//===----------------------------------------------------------------------===//
import NIO
#if compiler(>=5.5)
import _Concurrency
extension EventLoopFuture {
/// Get the value/error from an `EventLoopFuture` in an `async` context.
///
/// This function can be used to bridge an `EventLoopFuture` into the `async` world. Ie. if you're in an `async`
/// function and want to get the result of this future.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func get() async throws -> Value {
return try await withUnsafeThrowingContinuation { cont in
self.whenComplete { result in
switch result {
case .success(let value):
cont.resume(returning: value)
case .failure(let error):
cont.resume(throwing: error)
}
}
}
}
}
extension EventLoopGroup {
/// Shuts down the event loop gracefully.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func shutdownGracefully() async throws {
return try await withCheckedThrowingContinuation { cont in
self.shutdownGracefully { error in
if let error = error {
cont.resume(throwing: error)
} else {
cont.resume()
}
}
}
}
}
extension EventLoopPromise {
/// Complete a future with the result (or error) of the `async` function `body`.
///
/// This function can be used to bridge the `async` world into an `EventLoopPromise`.
///
/// - parameters:
/// - body: The `async` function to run.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func completeWithAsync(_ body: @escaping () async throws -> Value) {
Task {
do {
let value = try await body()
self.succeed(value)
} catch {
self.fail(error)
}
}
}
}
extension Channel {
/// Shortcut for calling `write` and `flush`.
///
/// - parameters:
/// - data: the data to write
/// - promise: the `EventLoopPromise` that will be notified once the `write` operation completes,
/// or `nil` if not interested in the outcome of the operation.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func writeAndFlush<T>(_ any: T) async throws {
try await self.writeAndFlush(any).get()
}
/// Set `option` to `value` on this `Channel`.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func setOption<Option: ChannelOption>(_ option: Option, value: Option.Value) async throws {
try await self.setOption(option, value: value).get()
}
/// Get the value of `option` for this `Channel`.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func getOption<Option: ChannelOption>(_ option: Option) async throws -> Option.Value {
return try await self.getOption(option).get()
}
}
extension ChannelOutboundInvoker {
/// Register on an `EventLoop` and so have all its IO handled.
///
/// - returns: the future which will be notified once the operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func register(file: StaticString = #file, line: UInt = #line) async throws {
try await self.register(file: file, line: line).get()
}
/// Bind to a `SocketAddress`.
/// - parameters:
/// - to: the `SocketAddress` to which we should bind the `Channel`.
/// - returns: the future which will be notified once the operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func bind(to address: SocketAddress, file: StaticString = #file, line: UInt = #line) async throws {
try await self.bind(to: address, file: file, line: line).get()
}
/// Connect to a `SocketAddress`.
/// - parameters:
/// - to: the `SocketAddress` to which we should connect the `Channel`.
/// - returns: the future which will be notified once the operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func connect(to address: SocketAddress, file: StaticString = #file, line: UInt = #line) async throws {
try await self.connect(to: address, file: file, line: line).get()
}
/// Shortcut for calling `write` and `flush`.
///
/// - parameters:
/// - data: the data to write
/// - returns: the future which will be notified once the `write` operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func writeAndFlush(_ data: NIOAny, file: StaticString = #file, line: UInt = #line) async throws {
try await self.writeAndFlush(data, file: file, line: line).get()
}
/// Close the `Channel` and so the connection if one exists.
///
/// - parameters:
/// - mode: the `CloseMode` that is used
/// - returns: the future which will be notified once the operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func close(mode: CloseMode = .all, file: StaticString = #file, line: UInt = #line) async throws {
try await self.close(mode: mode, file: file, line: line).get()
}
/// Trigger a custom user outbound event which will flow through the `ChannelPipeline`.
///
/// - parameters:
/// - event: the event itself.
/// - returns: the future which will be notified once the operation completes.
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func triggerUserOutboundEvent(_ event: Any, file: StaticString = #file, line: UInt = #line) async throws {
try await self.triggerUserOutboundEvent(event, file: file, line: line).get()
}
}
extension ChannelPipeline {
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func addHandler(_ handler: ChannelHandler,
name: String? = nil,
position: ChannelPipeline.Position = .last) async throws {
try await self.addHandler(handler, name: name, position: position).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func removeHandler(_ handler: RemovableChannelHandler) async throws {
try await self.removeHandler(handler).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func removeHandler(name: String) async throws {
try await self.removeHandler(name: name).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func removeHandler(context: ChannelHandlerContext) async throws {
try await self.removeHandler(context: context).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func context(handler: ChannelHandler) async throws -> ChannelHandlerContext {
return try await self.context(handler: handler).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func context(name: String) async throws -> ChannelHandlerContext {
return try await self.context(name: name).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
@inlinable
public func context<Handler: ChannelHandler>(handlerType: Handler.Type) async throws -> ChannelHandlerContext {
return try await self.context(handlerType: handlerType).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func addHandlers(_ handlers: [ChannelHandler],
position: ChannelPipeline.Position = .last) async throws {
try await self.addHandlers(handlers, position: position).get()
}
@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
public func addHandlers(_ handlers: ChannelHandler...,
position: ChannelPipeline.Position = .last) async throws {
try await self.addHandlers(handlers, position: position)
}
}
#endif
|
