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
|
// Copyright 2021 gRPC authors.
//
// 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.
#ifndef GRPC_SRC_CORE_EXT_TRANSPORT_BINDER_TRANSPORT_BINDER_STREAM_H
#define GRPC_SRC_CORE_EXT_TRANSPORT_BINDER_TRANSPORT_BINDER_STREAM_H
#include <grpc/support/port_platform.h>
#include "src/core/ext/transport/binder/transport/binder_transport.h"
struct RecvInitialMetadataArgs {
grpc_binder_stream* stream;
grpc_binder_transport* transport;
int tx_code;
absl::StatusOr<grpc_binder::Metadata> initial_metadata;
};
struct RecvMessageArgs {
grpc_binder_stream* stream;
grpc_binder_transport* transport;
int tx_code;
absl::StatusOr<std::string> message;
};
struct RecvTrailingMetadataArgs {
grpc_binder_stream* stream;
grpc_binder_transport* transport;
int tx_code;
absl::StatusOr<grpc_binder::Metadata> trailing_metadata;
int status;
};
struct RegisterStreamArgs {
grpc_binder_stream* stream;
grpc_binder_transport* transport;
};
// TODO(mingcl): Figure out if we want to use class instead of struct here
struct grpc_binder_stream {
// server_data will be null for client, and for server it will be whatever
// passed in to the accept_stream_fn callback by client.
grpc_binder_stream(grpc_binder_transport* t, grpc_stream_refcount* refcount,
const void* /*server_data*/, grpc_core::Arena* arena,
int tx_code, bool is_client)
: t(t),
refcount(refcount),
arena(arena),
tx_code(tx_code),
is_client(is_client),
is_closed(false) {
recv_initial_metadata_args.stream = this;
recv_initial_metadata_args.transport = t;
recv_message_args.stream = this;
recv_message_args.transport = t;
recv_trailing_metadata_args.stream = this;
recv_trailing_metadata_args.transport = t;
}
~grpc_binder_stream() {
if (destroy_stream_then_closure != nullptr) {
grpc_core::ExecCtx::Run(DEBUG_LOCATION, destroy_stream_then_closure,
absl::OkStatus());
}
}
int GetTxCode() const { return tx_code; }
grpc_binder_transport* t;
grpc_stream_refcount* refcount;
grpc_core::Arena* arena;
int tx_code;
const bool is_client;
bool is_closed;
grpc_closure* destroy_stream_then_closure = nullptr;
grpc_closure destroy_stream;
// The reason why this stream is cancelled and closed.
grpc_error_handle cancel_self_error;
grpc_closure recv_initial_metadata_closure;
RecvInitialMetadataArgs recv_initial_metadata_args;
grpc_closure recv_message_closure;
RecvMessageArgs recv_message_args;
grpc_closure recv_trailing_metadata_closure;
RecvTrailingMetadataArgs recv_trailing_metadata_args;
grpc_closure register_stream_closure;
RegisterStreamArgs register_stream_args;
// We store these fields passed from op batch, in order to access them through
// grpc_binder_stream
grpc_metadata_batch* recv_initial_metadata;
grpc_closure* recv_initial_metadata_ready = nullptr;
bool* trailing_metadata_available = nullptr;
absl::optional<grpc_core::SliceBuffer>* recv_message;
grpc_closure* recv_message_ready = nullptr;
bool* call_failed_before_recv_message = nullptr;
grpc_metadata_batch* recv_trailing_metadata;
grpc_closure* recv_trailing_metadata_finished = nullptr;
bool trailing_metadata_sent = false;
bool need_to_call_trailing_metadata_callback = false;
};
#endif // GRPC_SRC_CORE_EXT_TRANSPORT_BINDER_TRANSPORT_BINDER_STREAM_H
|
