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
|
/*
*
* Copyright 2014, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*
*/
#ifndef __GRPCPP_STREAM_H__
#define __GRPCPP_STREAM_H__
#include <grpc++/stream_context_interface.h>
#include <grpc++/status.h>
#include <grpc/support/log.h>
namespace grpc {
// Common interface for all client side streaming.
class ClientStreamingInterface {
public:
virtual ~ClientStreamingInterface() {}
// Try to cancel the stream. Wait() still needs to be called to get the final
// status. Cancelling after the stream has finished has no effects.
virtual void Cancel() = 0;
// Wait until the stream finishes, and return the final status. When the
// client side declares it has no more message to send, either implicitly or
// by calling WritesDone, it needs to make sure there is no more message to
// be received from the server, either implicitly or by getting a false from
// a Read(). Otherwise, this implicitly cancels the stream.
virtual const Status& Wait() = 0;
};
// An interface that yields a sequence of R messages.
template <class R>
class ReaderInterface {
public:
virtual ~ReaderInterface() {}
// Blocking read a message and parse to msg. Returns true on success.
// The method returns false when there will be no more incoming messages,
// either because the other side has called WritesDone or the stream has
// failed (or been cancelled).
virtual bool Read(R* msg) = 0;
};
// An interface that can be fed a sequence of W messages.
template <class W>
class WriterInterface {
public:
virtual ~WriterInterface() {}
// Blocking write msg to the stream. Returns true on success.
// Returns false when the stream has been closed.
virtual bool Write(const W& msg) = 0;
};
template <class R>
class ClientReader : public ClientStreamingInterface,
public ReaderInterface<R> {
public:
// Blocking create a stream and write the first request out.
explicit ClientReader(StreamContextInterface* context) : context_(context) {
GPR_ASSERT(context_);
context_->Start(true);
context_->Write(context_->request(), true);
}
~ClientReader() { delete context_; }
virtual bool Read(R* msg) { return context_->Read(msg); }
virtual void Cancel() { context_->FinishStream(Status::Cancelled, true); }
virtual const Status& Wait() { return context_->Wait(); }
private:
StreamContextInterface* const context_;
};
template <class W>
class ClientWriter : public ClientStreamingInterface,
public WriterInterface<W> {
public:
// Blocking create a stream.
explicit ClientWriter(StreamContextInterface* context) : context_(context) {
GPR_ASSERT(context_);
context_->Start(false);
}
~ClientWriter() { delete context_; }
virtual bool Write(const W& msg) {
return context_->Write(const_cast<W*>(&msg), false);
}
virtual void WritesDone() { context_->Write(nullptr, true); }
virtual void Cancel() { context_->FinishStream(Status::Cancelled, true); }
// Read the final response and wait for the final status.
virtual const Status& Wait() {
bool success = context_->Read(context_->response());
if (!success) {
Cancel();
} else {
success = context_->Read(nullptr);
if (success) {
Cancel();
}
}
return context_->Wait();
}
private:
StreamContextInterface* const context_;
};
// Client-side interface for bi-directional streaming.
template <class W, class R>
class ClientReaderWriter : public ClientStreamingInterface,
public WriterInterface<W>,
public ReaderInterface<R> {
public:
// Blocking create a stream.
explicit ClientReaderWriter(StreamContextInterface* context)
: context_(context) {
GPR_ASSERT(context_);
context_->Start(false);
}
~ClientReaderWriter() { delete context_; }
virtual bool Read(R* msg) { return context_->Read(msg); }
virtual bool Write(const W& msg) {
return context_->Write(const_cast<W*>(&msg), false);
}
virtual void WritesDone() { context_->Write(nullptr, true); }
virtual void Cancel() { context_->FinishStream(Status::Cancelled, true); }
virtual const Status& Wait() { return context_->Wait(); }
private:
StreamContextInterface* const context_;
};
} // namespace grpc
#endif // __GRPCPP_STREAM_H__
|
