Skip to content

Commit 7c1f9f2

Browse files
author
Rustin
authored
feat: add the WatchState API (#582)
This pull request introduces a new stream API named "WatchState." Currently, it only provides the current temporality, which will enable clients to check if the server has been paused.
1 parent 1f41b61 commit 7c1f9f2

File tree

9 files changed

+408
-48
lines changed

9 files changed

+408
-48
lines changed

console-api/proto/instrument.proto

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ service Instrument {
1414
rpc WatchUpdates(InstrumentRequest) returns (stream Update) {}
1515
// Produces a stream of updates describing the activity of a specific task.
1616
rpc WatchTaskDetails(TaskDetailsRequest) returns (stream tasks.TaskDetails) {}
17+
// Produces a stream of state of the aggregator.
18+
rpc WatchState(StateRequest) returns (stream State) {}
1719
// Registers that the console observer wants to pause the stream.
1820
rpc Pause(PauseRequest) returns (PauseResponse) {}
1921
// Registers that the console observer wants to resume the stream.
@@ -72,6 +74,23 @@ message Update {
7274
common.RegisterMetadata new_metadata = 5;
7375
}
7476

77+
// StateRequest requests the current state of the aggregator.
78+
message StateRequest {
79+
}
80+
81+
// State carries the current state of the aggregator.
82+
message State {
83+
Temporality temporality = 1;
84+
}
85+
86+
// The time "state" of the aggregator.
87+
enum Temporality {
88+
// The aggregator is currently live.
89+
LIVE = 0;
90+
// The aggregator is currently paused.
91+
PAUSED = 1;
92+
}
93+
7594
// `PauseResponse` is the value returned after a pause request.
7695
message PauseResponse {
7796
}

console-api/src/generated/rs.tokio.console.instrument.rs

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,12 +50,50 @@ pub struct Update {
5050
#[prost(message, optional, tag = "5")]
5151
pub new_metadata: ::core::option::Option<super::common::RegisterMetadata>,
5252
}
53+
/// StateRequest requests the current state of the aggregator.
54+
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
55+
pub struct StateRequest {}
56+
/// State carries the current state of the aggregator.
57+
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
58+
pub struct State {
59+
#[prost(enumeration = "Temporality", tag = "1")]
60+
pub temporality: i32,
61+
}
5362
/// `PauseResponse` is the value returned after a pause request.
5463
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
5564
pub struct PauseResponse {}
5665
/// `ResumeResponse` is the value returned after a resume request.
5766
#[derive(Clone, Copy, PartialEq, ::prost::Message)]
5867
pub struct ResumeResponse {}
68+
/// The time "state" of the aggregator.
69+
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
70+
#[repr(i32)]
71+
pub enum Temporality {
72+
/// The aggregator is currently live.
73+
Live = 0,
74+
/// The aggregator is currently paused.
75+
Paused = 1,
76+
}
77+
impl Temporality {
78+
/// String value of the enum field names used in the ProtoBuf definition.
79+
///
80+
/// The values are not transformed in any way and thus are considered stable
81+
/// (if the ProtoBuf definition does not change) and safe for programmatic use.
82+
pub fn as_str_name(&self) -> &'static str {
83+
match self {
84+
Self::Live => "LIVE",
85+
Self::Paused => "PAUSED",
86+
}
87+
}
88+
/// Creates an enum from field names used in the ProtoBuf definition.
89+
pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
90+
match value {
91+
"LIVE" => Some(Self::Live),
92+
"PAUSED" => Some(Self::Paused),
93+
_ => None,
94+
}
95+
}
96+
}
5997
/// Generated client implementations.
6098
pub mod instrument_client {
6199
#![allow(
@@ -208,6 +246,36 @@ pub mod instrument_client {
208246
);
209247
self.inner.server_streaming(req, path, codec).await
210248
}
249+
/// Produces a stream of state of the aggregator.
250+
pub async fn watch_state(
251+
&mut self,
252+
request: impl tonic::IntoRequest<super::StateRequest>,
253+
) -> std::result::Result<
254+
tonic::Response<tonic::codec::Streaming<super::State>>,
255+
tonic::Status,
256+
> {
257+
self.inner
258+
.ready()
259+
.await
260+
.map_err(|e| {
261+
tonic::Status::unknown(
262+
format!("Service was not ready: {}", e.into()),
263+
)
264+
})?;
265+
let codec = tonic::codec::ProstCodec::default();
266+
let path = http::uri::PathAndQuery::from_static(
267+
"/rs.tokio.console.instrument.Instrument/WatchState",
268+
);
269+
let mut req = request.into_request();
270+
req.extensions_mut()
271+
.insert(
272+
GrpcMethod::new(
273+
"rs.tokio.console.instrument.Instrument",
274+
"WatchState",
275+
),
276+
);
277+
self.inner.server_streaming(req, path, codec).await
278+
}
211279
/// Registers that the console observer wants to pause the stream.
212280
pub async fn pause(
213281
&mut self,
@@ -302,6 +370,17 @@ pub mod instrument_server {
302370
tonic::Response<Self::WatchTaskDetailsStream>,
303371
tonic::Status,
304372
>;
373+
/// Server streaming response type for the WatchState method.
374+
type WatchStateStream: tonic::codegen::tokio_stream::Stream<
375+
Item = std::result::Result<super::State, tonic::Status>,
376+
>
377+
+ std::marker::Send
378+
+ 'static;
379+
/// Produces a stream of state of the aggregator.
380+
async fn watch_state(
381+
&self,
382+
request: tonic::Request<super::StateRequest>,
383+
) -> std::result::Result<tonic::Response<Self::WatchStateStream>, tonic::Status>;
305384
/// Registers that the console observer wants to pause the stream.
306385
async fn pause(
307386
&self,
@@ -482,6 +561,52 @@ pub mod instrument_server {
482561
};
483562
Box::pin(fut)
484563
}
564+
"/rs.tokio.console.instrument.Instrument/WatchState" => {
565+
#[allow(non_camel_case_types)]
566+
struct WatchStateSvc<T: Instrument>(pub Arc<T>);
567+
impl<
568+
T: Instrument,
569+
> tonic::server::ServerStreamingService<super::StateRequest>
570+
for WatchStateSvc<T> {
571+
type Response = super::State;
572+
type ResponseStream = T::WatchStateStream;
573+
type Future = BoxFuture<
574+
tonic::Response<Self::ResponseStream>,
575+
tonic::Status,
576+
>;
577+
fn call(
578+
&mut self,
579+
request: tonic::Request<super::StateRequest>,
580+
) -> Self::Future {
581+
let inner = Arc::clone(&self.0);
582+
let fut = async move {
583+
<T as Instrument>::watch_state(&inner, request).await
584+
};
585+
Box::pin(fut)
586+
}
587+
}
588+
let accept_compression_encodings = self.accept_compression_encodings;
589+
let send_compression_encodings = self.send_compression_encodings;
590+
let max_decoding_message_size = self.max_decoding_message_size;
591+
let max_encoding_message_size = self.max_encoding_message_size;
592+
let inner = self.inner.clone();
593+
let fut = async move {
594+
let method = WatchStateSvc(inner);
595+
let codec = tonic::codec::ProstCodec::default();
596+
let mut grpc = tonic::server::Grpc::new(codec)
597+
.apply_compression_config(
598+
accept_compression_encodings,
599+
send_compression_encodings,
600+
)
601+
.apply_max_message_size_config(
602+
max_decoding_message_size,
603+
max_encoding_message_size,
604+
);
605+
let res = grpc.server_streaming(method, req).await;
606+
Ok(res)
607+
};
608+
Box::pin(fut)
609+
}
485610
"/rs.tokio.console.instrument.Instrument/Pause" => {
486611
#[allow(non_camel_case_types)]
487612
struct PauseSvc<T: Instrument>(pub Arc<T>);

console-subscriber/examples/grpc_web/app/src/gen/instrument_connect.ts

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
/* eslint-disable */
44
// @ts-nocheck
55

6-
import { InstrumentRequest, PauseRequest, PauseResponse, ResumeRequest, ResumeResponse, TaskDetailsRequest, Update } from "./instrument_pb.js";
6+
import { InstrumentRequest, PauseRequest, PauseResponse, ResumeRequest, ResumeResponse, State, StateRequest, TaskDetailsRequest, Update } from "./instrument_pb.js";
77
import { MethodKind } from "@bufbuild/protobuf";
88
import { TaskDetails } from "./tasks_pb.js";
99

@@ -37,6 +37,17 @@ export const Instrument = {
3737
O: TaskDetails,
3838
kind: MethodKind.ServerStreaming,
3939
},
40+
/**
41+
* Produces a stream of state of the aggregator.
42+
*
43+
* @generated from rpc rs.tokio.console.instrument.Instrument.WatchState
44+
*/
45+
watchState: {
46+
name: "WatchState",
47+
I: StateRequest,
48+
O: State,
49+
kind: MethodKind.ServerStreaming,
50+
},
4051
/**
4152
* Registers that the console observer wants to pause the stream.
4253
*

console-subscriber/examples/grpc_web/app/src/gen/instrument_pb.ts

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,32 @@ import { TaskUpdate } from "./tasks_pb.js";
1010
import { ResourceUpdate } from "./resources_pb.js";
1111
import { AsyncOpUpdate } from "./async_ops_pb.js";
1212

13+
/**
14+
* The time "state" of the aggregator.
15+
*
16+
* @generated from enum rs.tokio.console.instrument.Temporality
17+
*/
18+
export enum Temporality {
19+
/**
20+
* The aggregator is currently live.
21+
*
22+
* @generated from enum value: LIVE = 0;
23+
*/
24+
LIVE = 0,
25+
26+
/**
27+
* The aggregator is currently paused.
28+
*
29+
* @generated from enum value: PAUSED = 1;
30+
*/
31+
PAUSED = 1,
32+
}
33+
// Retrieve enum metadata with: proto3.getEnumType(Temporality)
34+
proto3.util.setEnumType(Temporality, "rs.tokio.console.instrument.Temporality", [
35+
{ no: 0, name: "LIVE" },
36+
{ no: 1, name: "PAUSED" },
37+
]);
38+
1339
/**
1440
* InstrumentRequest requests the stream of updates
1541
* to observe the async runtime state over time.
@@ -239,6 +265,78 @@ export class Update extends Message<Update> {
239265
}
240266
}
241267

268+
/**
269+
* StateRequest requests the current state of the aggregator.
270+
*
271+
* @generated from message rs.tokio.console.instrument.StateRequest
272+
*/
273+
export class StateRequest extends Message<StateRequest> {
274+
constructor(data?: PartialMessage<StateRequest>) {
275+
super();
276+
proto3.util.initPartial(data, this);
277+
}
278+
279+
static readonly runtime: typeof proto3 = proto3;
280+
static readonly typeName = "rs.tokio.console.instrument.StateRequest";
281+
static readonly fields: FieldList = proto3.util.newFieldList(() => [
282+
]);
283+
284+
static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): StateRequest {
285+
return new StateRequest().fromBinary(bytes, options);
286+
}
287+
288+
static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): StateRequest {
289+
return new StateRequest().fromJson(jsonValue, options);
290+
}
291+
292+
static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): StateRequest {
293+
return new StateRequest().fromJsonString(jsonString, options);
294+
}
295+
296+
static equals(a: StateRequest | PlainMessage<StateRequest> | undefined, b: StateRequest | PlainMessage<StateRequest> | undefined): boolean {
297+
return proto3.util.equals(StateRequest, a, b);
298+
}
299+
}
300+
301+
/**
302+
* State carries the current state of the aggregator.
303+
*
304+
* @generated from message rs.tokio.console.instrument.State
305+
*/
306+
export class State extends Message<State> {
307+
/**
308+
* @generated from field: rs.tokio.console.instrument.Temporality temporality = 1;
309+
*/
310+
temporality = Temporality.LIVE;
311+
312+
constructor(data?: PartialMessage<State>) {
313+
super();
314+
proto3.util.initPartial(data, this);
315+
}
316+
317+
static readonly runtime: typeof proto3 = proto3;
318+
static readonly typeName = "rs.tokio.console.instrument.State";
319+
static readonly fields: FieldList = proto3.util.newFieldList(() => [
320+
{ no: 1, name: "temporality", kind: "enum", T: proto3.getEnumType(Temporality) },
321+
]);
322+
323+
static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): State {
324+
return new State().fromBinary(bytes, options);
325+
}
326+
327+
static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): State {
328+
return new State().fromJson(jsonValue, options);
329+
}
330+
331+
static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): State {
332+
return new State().fromJsonString(jsonString, options);
333+
}
334+
335+
static equals(a: State | PlainMessage<State> | undefined, b: State | PlainMessage<State> | undefined): boolean {
336+
return proto3.util.equals(State, a, b);
337+
}
338+
}
339+
242340
/**
243341
* `PauseResponse` is the value returned after a pause request.
244342
*

0 commit comments

Comments
 (0)