@@ -43,14 +43,14 @@ internal constructor(
43
43
override suspend fun <T : Any > get (key : StateKey <T >): T ? =
44
44
resolveSerde<T ?>(key.serdeInfo())
45
45
.let { serde ->
46
- SingleAwaitableImpl (handlerContext.get(key.name()).await()).simpleMap {
46
+ SingleDurableFutureImpl (handlerContext.get(key.name()).await()).simpleMap {
47
47
it.getOrNull()?.let { serde.deserialize(it) }
48
48
}
49
49
}
50
50
.await()
51
51
52
52
override suspend fun stateKeys (): Collection <String > =
53
- SingleAwaitableImpl (handlerContext.getKeys().await()).await()
53
+ SingleDurableFutureImpl (handlerContext.getKeys().await()).await()
54
54
55
55
override suspend fun <T : Any > set (key : StateKey <T >, value : T ) {
56
56
handlerContext.set(key.name(), resolveAndSerialize(key.serdeInfo(), value)).await()
@@ -64,12 +64,12 @@ internal constructor(
64
64
handlerContext.clearAll().await()
65
65
}
66
66
67
- override suspend fun timer (duration : Duration , name : String? ): Awaitable <Unit > =
68
- SingleAwaitableImpl (handlerContext.timer(duration.toJavaDuration(), name).await()).map {}
67
+ override suspend fun timer (duration : Duration , name : String? ): DurableFuture <Unit > =
68
+ SingleDurableFutureImpl (handlerContext.timer(duration.toJavaDuration(), name).await()).map {}
69
69
70
70
override suspend fun <Req : Any ?, Res : Any ?> call (
71
71
request : Request <Req , Res >
72
- ): CallAwaitable <Res > =
72
+ ): CallDurableFuture <Res > =
73
73
resolveSerde<Res >(request.responseTypeTag()).let { responseSerde ->
74
74
val callHandle =
75
75
handlerContext
@@ -85,7 +85,7 @@ internal constructor(
85
85
CompletableFuture .completedFuture<Res >(responseSerde.deserialize(it))
86
86
}
87
87
88
- return @let CallAwaitableImpl (callAsyncResult, callHandle.invocationIdAsyncResult)
88
+ return @let CallDurableFutureImpl (callAsyncResult, callHandle.invocationIdAsyncResult)
89
89
}
90
90
91
91
override suspend fun <Req : Any ?, Res : Any ?> send (
@@ -122,7 +122,7 @@ internal constructor(
122
122
name : String ,
123
123
retryPolicy : RetryPolicy ? ,
124
124
block : suspend () -> T
125
- ): Awaitable <T > {
125
+ ): DurableFuture <T > {
126
126
var serde: Serde <T > = resolveSerde(typeTag)
127
127
var coroutineCtx = currentCoroutineContext()
128
128
val javaRetryPolicy =
@@ -151,7 +151,7 @@ internal constructor(
151
151
}
152
152
}
153
153
.await()
154
- return SingleAwaitableImpl (asyncResult).map { serde.deserialize(it) }
154
+ return SingleDurableFutureImpl (asyncResult).map { serde.deserialize(it) }
155
155
}
156
156
157
157
override suspend fun <T : Any > awakeable (typeTag : TypeTag <T >): Awakeable <T > {
@@ -180,13 +180,13 @@ internal constructor(
180
180
DurablePromise <T > {
181
181
val serde: Serde <T > = resolveSerde(key.serdeInfo())
182
182
183
- override suspend fun awaitable (): Awaitable <T > =
184
- SingleAwaitableImpl (handlerContext.promise(key.name()).await()).simpleMap {
183
+ override suspend fun future (): DurableFuture <T > =
184
+ SingleDurableFutureImpl (handlerContext.promise(key.name()).await()).simpleMap {
185
185
serde.deserialize(it)
186
186
}
187
187
188
188
override suspend fun peek (): Output <T > =
189
- SingleAwaitableImpl (handlerContext.peekPromise(key.name()).await())
189
+ SingleDurableFutureImpl (handlerContext.peekPromise(key.name()).await())
190
190
.simpleMap { it.map { serde.deserialize(it) } }
191
191
.await()
192
192
}
@@ -196,7 +196,7 @@ internal constructor(
196
196
val serde: Serde <T > = resolveSerde(key.serdeInfo())
197
197
198
198
override suspend fun resolve (payload : T ) {
199
- SingleAwaitableImpl (
199
+ SingleDurableFutureImpl (
200
200
handlerContext
201
201
.resolvePromise(
202
202
key.name(), serde.serializeWrappingException(handlerContext, payload))
@@ -205,7 +205,7 @@ internal constructor(
205
205
}
206
206
207
207
override suspend fun reject (reason : String ) {
208
- SingleAwaitableImpl (
208
+ SingleDurableFutureImpl (
209
209
handlerContext.rejectPromise(key.name(), TerminalException (reason)).await())
210
210
.await()
211
211
}
0 commit comments