@ -30,7 +30,6 @@
namespace grpc {
namespace internal {
/// Common interface for all synchronous client side streaming.
class ClientStreamingInterface {
public :
@ -63,6 +62,20 @@ class ClientStreamingInterface {
virtual Status Finish ( ) = 0 ;
} ;
/// Common interface for all synchronous server side streaming.
class ServerStreamingInterface {
public :
virtual ~ ServerStreamingInterface ( ) { }
/// Block to send initial metadata to client.
/// This call is optional, but if it is used, it cannot be used concurrently
/// with or after the \a Finish method.
///
/// The initial metadata that will be sent to the client will be
/// taken from the \a ServerContext associated with the call.
virtual void SendInitialMetadata ( ) = 0 ;
} ;
/// An interface that yields a sequence of messages of type \a R.
template < class R >
class ReaderInterface {
@ -128,55 +141,16 @@ class WriterInterface {
}
} ;
} // namespace internal
/// Client-side interface for streaming reads of message of type \a R.
template < class R >
class ClientReaderInterface : public internal : : ClientStreamingInterface ,
public internal : : ReaderInterface < R > {
public :
/// Block to wait for initial metadata from server. The received metadata
/// can only be accessed after this call returns. Should only be called before
/// the first read. Calling this method is optional, and if it is not called
/// the metadata will be available in ClientContext after the first read.
virtual void WaitForInitialMetadata ( ) = 0 ;
} ;
/// Client-side interface for streaming writes of message type \a W.
template < class W >
class ClientWriterInterface : public internal : : ClientStreamingInterface ,
public internal : : WriterInterface < W > {
public :
/// Half close writing from the client. (signal that the stream of messages
/// coming from the clinet is complete).
/// Blocks until currently-pending writes are completed.
/// Thread safe with respect to \a ReaderInterface::Read operations only
///
/// \return Whether the writes were successful.
virtual bool WritesDone ( ) = 0 ;
} ;
/// Client-side interface for bi-directional streaming with
/// client-to-server stream messages of type \a W and
/// server-to-client stream messages of type \a R.
template < class W , class R >
class ClientReaderWriterInterface : public internal : : ClientStreamingInterface ,
public internal : : WriterInterface < W > ,
public internal : : ReaderInterface < R > {
class ClientReaderInterface : public ClientStreamingInterface ,
public ReaderInterface < R > {
public :
/// Block to wait for initial metadata from server. The received metadata
/// can only be accessed after this call returns. Should only be called before
/// the first read. Calling this method is optional, and if it is not called
/// the metadata will be available in ClientContext after the first read.
virtual void WaitForInitialMetadata ( ) = 0 ;
/// Half close writing from the client. (signal that the stream of messages
/// coming from the clinet is complete).
/// Blocks until currently-pending writes are completed.
/// Thread-safe with respect to \a ReaderInterface::Read
///
/// \return Whether the writes were successful.
virtual bool WritesDone ( ) = 0 ;
} ;
/// Synchronous (blocking) client-side API for doing server-streaming RPCs,
@ -185,14 +159,28 @@ class ClientReaderWriterInterface : public internal::ClientStreamingInterface,
template < class R >
class ClientReader final : public ClientReaderInterface < R > {
public :
struct internal {
template < class W >
static ClientReader * Create ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context , const W & request ) {
return new ClientReader ( channel , method , context , request ) ;
}
} ;
/// Block to create a stream and write the initial metadata and \a request
/// out. Note that \a context will be used to fill in custom initial
/// metadata used to send to the server when starting the call.
template < class W >
ClientReader ( ChannelInterface * channel , const RpcMethod & method ,
ClientContext * context , const W & request )
: context_ ( context ) ,
cq_ ( grpc_completion_queue_attributes {
GRPC_CQ_CURRENT_VERSION , GRPC_CQ_PLUCK ,
GRPC_CQ_DEFAULT_POLLING } ) , // Pluckable cq
call_ ( channel - > CreateCall ( method , context , & cq_ ) ) {
CallOpSet < CallOpSendInitialMetadata , CallOpSendMessage ,
CallOpClientSendClose >
ops ;
ops . SendInitialMetadata ( context - > send_initial_metadata_ ,
context - > initial_metadata_flags ( ) ) ;
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT ( ops . SendMessage ( request ) . ok ( ) ) ;
ops . ClientSendClose ( ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ;
}
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
/// semantics.
@ -204,8 +192,7 @@ class ClientReader final : public ClientReaderInterface<R> {
void WaitForInitialMetadata ( ) override {
GPR_CODEGEN_ASSERT ( ! context_ - > initial_metadata_received_ ) ;
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata >
ops ;
CallOpSet < CallOpRecvInitialMetadata > ops ;
ops . RecvInitialMetadata ( context_ ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ; /// status ignored
@ -222,9 +209,7 @@ class ClientReader final : public ClientReaderInterface<R> {
/// already received (if initial metadata is received, it can be then
/// accessed through the \a ClientContext associated with this call).
bool Read ( R * msg ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata ,
: : grpc : : internal : : CallOpRecvMessage < R > >
ops ;
CallOpSet < CallOpRecvInitialMetadata , CallOpRecvMessage < R > > ops ;
if ( ! context_ - > initial_metadata_received_ ) {
ops . RecvInitialMetadata ( context_ ) ;
}
@ -239,7 +224,7 @@ class ClientReader final : public ClientReaderInterface<R> {
/// The \a ClientContext associated with this call is updated with
/// possible metadata received from the server.
Status Finish ( ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpClientRecvStatus > ops ;
CallOpSet < CallOpClientRecvStatus > ops ;
Status status ;
ops . ClientRecvStatus ( context_ , & status ) ;
call_ . PerformOps ( & ops ) ;
@ -250,48 +235,53 @@ class ClientReader final : public ClientReaderInterface<R> {
private :
ClientContext * context_ ;
CompletionQueue cq_ ;
: : grpc : : internal : : Call call_ ;
Call call_ ;
} ;
/// Block to create a stream and write the initial metadata and \a request
/// out. Note that \a context will be used to fill in custom initial
/// metadata used to send to the server when starting the call.
template < class W >
ClientReader ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context , const W & request )
: context_ ( context ) ,
cq_ ( grpc_completion_queue_attributes {
GRPC_CQ_CURRENT_VERSION , GRPC_CQ_PLUCK ,
GRPC_CQ_DEFAULT_POLLING } ) , // Pluckable cq
call_ ( channel - > CreateCall ( method , context , & cq_ ) ) {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpSendInitialMetadata ,
: : grpc : : internal : : CallOpSendMessage ,
: : grpc : : internal : : CallOpClientSendClose >
ops ;
ops . SendInitialMetadata ( context - > send_initial_metadata_ ,
context - > initial_metadata_flags ( ) ) ;
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT ( ops . SendMessage ( request ) . ok ( ) ) ;
ops . ClientSendClose ( ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ;
}
/// Client-side interface for streaming writes of message type \a W.
template < class W >
class ClientWriterInterface : public ClientStreamingInterface ,
public WriterInterface < W > {
public :
/// Half close writing from the client. (signal that the stream of messages
/// coming from the clinet is complete).
/// Blocks until currently-pending writes are completed.
/// Thread safe with respect to \a ReaderInterface::Read operations only
///
/// \return Whether the writes were successful.
virtual bool WritesDone ( ) = 0 ;
} ;
/// Synchronous (blocking) client-side API for doing client-streaming RPCs,
/// where the outgoing message stream coming from the client has messages of
/// type \a W.
template < class W >
class ClientWriter final : public ClientWriterInterface < W > {
class ClientWriter : public ClientWriterInterface < W > {
public :
struct internal {
template < class R >
static ClientWriter * Create ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context , R * response ) {
return new ClientWriter ( channel , method , context , response ) ;
/// Block to create a stream (i.e. send request headers and other initial
/// metadata to the server). Note that \a context will be used to fill
/// in custom initial metadata. \a response will be filled in with the
/// single expected response message from the server upon a successful
/// call to the \a Finish method of this instance.
template < class R >
ClientWriter ( ChannelInterface * channel , const RpcMethod & method ,
ClientContext * context , R * response )
: context_ ( context ) ,
cq_ ( grpc_completion_queue_attributes {
GRPC_CQ_CURRENT_VERSION , GRPC_CQ_PLUCK ,
GRPC_CQ_DEFAULT_POLLING } ) , // Pluckable cq
call_ ( channel - > CreateCall ( method , context , & cq_ ) ) {
finish_ops_ . RecvMessage ( response ) ;
finish_ops_ . AllowNoMessage ( ) ;
if ( ! context_ - > initial_metadata_corked_ ) {
CallOpSet < CallOpSendInitialMetadata > ops ;
ops . SendInitialMetadata ( context - > send_initial_metadata_ ,
context - > initial_metadata_flags ( ) ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ;
}
} ;
}
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
/// semantics.
@ -302,8 +292,7 @@ class ClientWriter final : public ClientWriterInterface<W> {
void WaitForInitialMetadata ( ) {
GPR_CODEGEN_ASSERT ( ! context_ - > initial_metadata_received_ ) ;
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata >
ops ;
CallOpSet < CallOpRecvInitialMetadata > ops ;
ops . RecvInitialMetadata ( context_ ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ; // status ignored
@ -315,11 +304,10 @@ class ClientWriter final : public ClientWriterInterface<W> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the
/// \a ClientContext associated with this call).
using : : grpc : : internal : : WriterInterface < W > : : Write ;
using WriterInterface < W > : : Write ;
bool Write ( const W & msg , WriteOptions options ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpSendInitialMetadata ,
: : grpc : : internal : : CallOpSendMessage ,
: : grpc : : internal : : CallOpClientSendClose >
CallOpSet < CallOpSendInitialMetadata , CallOpSendMessage ,
CallOpClientSendClose >
ops ;
if ( options . is_last_message ( ) ) {
@ -340,7 +328,7 @@ class ClientWriter final : public ClientWriterInterface<W> {
}
bool WritesDone ( ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpClientSendClose > ops ;
CallOpSet < CallOpClientSendClose > ops ;
ops . ClientSendClose ( ) ;
call_ . PerformOps ( & ops ) ;
return cq_ . Pluck ( & ops ) ;
@ -365,55 +353,61 @@ class ClientWriter final : public ClientWriterInterface<W> {
private :
ClientContext * context_ ;
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata ,
: : grpc : : internal : : CallOpGenericRecvMessage ,
: : grpc : : internal : : CallOpClientRecvStatus >
CallOpSet < CallOpRecvInitialMetadata , CallOpGenericRecvMessage ,
CallOpClientRecvStatus >
finish_ops_ ;
CompletionQueue cq_ ;
: : grpc : : internal : : Call call_ ;
Call call_ ;
} ;
/// Block to create a stream (i.e. send request headers and other initial
/// metadata to the server). Note that \a context will be used to fill
/// in custom initial metadata. \a response will be filled in with the
/// single expected response message from the server upon a successful
/// call to the \a Finish method of this instance.
template < class R >
ClientWriter ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context , R * response )
/// Client-side interface for bi-directional streaming with
/// client-to-server stream messages of type \a W and
/// server-to-client stream messages of type \a R.
template < class W , class R >
class ClientReaderWriterInterface : public ClientStreamingInterface ,
public WriterInterface < W > ,
public ReaderInterface < R > {
public :
/// Block to wait for initial metadata from server. The received metadata
/// can only be accessed after this call returns. Should only be called before
/// the first read. Calling this method is optional, and if it is not called
/// the metadata will be available in ClientContext after the first read.
virtual void WaitForInitialMetadata ( ) = 0 ;
/// Half close writing from the client. (signal that the stream of messages
/// coming from the clinet is complete).
/// Blocks until currently-pending writes are completed.
/// Thread-safe with respect to \a ReaderInterface::Read
///
/// \return Whether the writes were successful.
virtual bool WritesDone ( ) = 0 ;
} ;
/// Synchronous (blocking) client-side API for bi-directional streaming RPCs,
/// where the outgoing message stream coming from the client has messages of
/// type \a W, and the incoming messages stream coming from the server has
/// messages of type \a R.
template < class W , class R >
class ClientReaderWriter final : public ClientReaderWriterInterface < W , R > {
public :
/// Block to create a stream and write the initial metadata and \a request
/// out. Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
ClientReaderWriter ( ChannelInterface * channel , const RpcMethod & method ,
ClientContext * context )
: context_ ( context ) ,
cq_ ( grpc_completion_queue_attributes {
GRPC_CQ_CURRENT_VERSION , GRPC_CQ_PLUCK ,
GRPC_CQ_DEFAULT_POLLING } ) , // Pluckable cq
call_ ( channel - > CreateCall ( method , context , & cq_ ) ) {
finish_ops_ . RecvMessage ( response ) ;
finish_ops_ . AllowNoMessage ( ) ;
if ( ! context_ - > initial_metadata_corked_ ) {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpSendInitialMetadata >
ops ;
CallOpSet < CallOpSendInitialMetadata > ops ;
ops . SendInitialMetadata ( context - > send_initial_metadata_ ,
context - > initial_metadata_flags ( ) ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ;
}
}
} ;
/// Synchronous (blocking) client-side API for bi-directional streaming RPCs,
/// where the outgoing message stream coming from the client has messages of
/// type \a W, and the incoming messages stream coming from the server has
/// messages of type \a R.
template < class W , class R >
class ClientReaderWriter final : public ClientReaderWriterInterface < W , R > {
public :
struct internal {
static ClientReaderWriter * Create ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context ) {
return new ClientReaderWriter ( channel , method , context ) ;
}
} ;
/// Block waiting to read initial metadata from the server.
/// This call is optional, but if it is used, it cannot be used concurrently
@ -424,8 +418,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
void WaitForInitialMetadata ( ) override {
GPR_CODEGEN_ASSERT ( ! context_ - > initial_metadata_received_ ) ;
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata >
ops ;
CallOpSet < CallOpRecvInitialMetadata > ops ;
ops . RecvInitialMetadata ( context_ ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ; // status ignored
@ -441,9 +434,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// Also receives initial metadata if not already received (updates the \a
/// ClientContext associated with this call in that case).
bool Read ( R * msg ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata ,
: : grpc : : internal : : CallOpRecvMessage < R > >
ops ;
CallOpSet < CallOpRecvInitialMetadata , CallOpRecvMessage < R > > ops ;
if ( ! context_ - > initial_metadata_received_ ) {
ops . RecvInitialMetadata ( context_ ) ;
}
@ -457,11 +448,10 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the
/// \a ClientContext associated with this call to fill in values).
using : : grpc : : internal : : WriterInterface < W > : : Write ;
using WriterInterface < W > : : Write ;
bool Write ( const W & msg , WriteOptions options ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpSendInitialMetadata ,
: : grpc : : internal : : CallOpSendMessage ,
: : grpc : : internal : : CallOpClientSendClose >
CallOpSet < CallOpSendInitialMetadata , CallOpSendMessage ,
CallOpClientSendClose >
ops ;
if ( options . is_last_message ( ) ) {
@ -482,7 +472,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
}
bool WritesDone ( ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpClientSendClose > ops ;
CallOpSet < CallOpClientSendClose > ops ;
ops . ClientSendClose ( ) ;
call_ . PerformOps ( & ops ) ;
return cq_ . Pluck ( & ops ) ;
@ -494,9 +484,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// - the \a ClientContext associated with this call is updated with
/// possible trailing metadata sent from the server.
Status Finish ( ) override {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpRecvInitialMetadata ,
: : grpc : : internal : : CallOpClientRecvStatus >
ops ;
CallOpSet < CallOpRecvInitialMetadata , CallOpClientRecvStatus > ops ;
if ( ! context_ - > initial_metadata_received_ ) {
ops . RecvInitialMetadata ( context_ ) ;
}
@ -510,61 +498,13 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
private :
ClientContext * context_ ;
CompletionQueue cq_ ;
: : grpc : : internal : : Call call_ ;
/// Block to create a stream and write the initial metadata and \a request
/// out. Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
ClientReaderWriter ( : : grpc : : ChannelInterface * channel ,
const : : grpc : : internal : : RpcMethod & method ,
ClientContext * context )
: context_ ( context ) ,
cq_ ( grpc_completion_queue_attributes {
GRPC_CQ_CURRENT_VERSION , GRPC_CQ_PLUCK ,
GRPC_CQ_DEFAULT_POLLING } ) , // Pluckable cq
call_ ( channel - > CreateCall ( method , context , & cq_ ) ) {
if ( ! context_ - > initial_metadata_corked_ ) {
: : grpc : : internal : : CallOpSet < : : grpc : : internal : : CallOpSendInitialMetadata >
ops ;
ops . SendInitialMetadata ( context - > send_initial_metadata_ ,
context - > initial_metadata_flags ( ) ) ;
call_ . PerformOps ( & ops ) ;
cq_ . Pluck ( & ops ) ;
}
}
Call call_ ;
} ;
namespace internal {
/// Common interface for all synchronous server side streaming.
class ServerStreamingInterface {
public :
virtual ~ ServerStreamingInterface ( ) { }
/// Block to send initial metadata to client.
/// This call is optional, but if it is used, it cannot be used concurrently
/// with or after the \a Finish method.
///
/// The initial metadata that will be sent to the client will be
/// taken from the \a ServerContext associated with the call.
virtual void SendInitialMetadata ( ) = 0 ;
} ;
} // namespace internal
/// Server-side interface for streaming reads of message of type \a R.
template < class R >
class ServerReaderInterface : public internal : : ServerStreamingInterface ,
public internal : : ReaderInterface < R > { } ;
/// Server-side interface for streaming writes of message of type \a W.
template < class W >
class ServerWriterInterface : public internal : : ServerStreamingInterface ,
public internal : : WriterInterface < W > { } ;
/// Server-side interface for bi-directional streaming.
template < class W , class R >
class ServerReaderWriterInterface : public internal : : ServerStreamingInterface ,
public internal : : WriterInterface < W > ,
public internal : : ReaderInterface < R > { } ;
class ServerReaderInterface : public ServerStreamingInterface ,
public ReaderInterface < R > { } ;
/// Synchronous (blocking) server-side API for doing client-streaming RPCs,
/// where the incoming message stream coming from the client has messages of
@ -572,13 +512,15 @@ class ServerReaderWriterInterface : public internal::ServerStreamingInterface,
template < class R >
class ServerReader final : public ServerReaderInterface < R > {
public :
ServerReader ( Call * call , ServerContext * ctx ) : call_ ( call ) , ctx_ ( ctx ) { }
/// See the \a ServerStreamingInterface.SendInitialMetadata method
/// for semantics. Note that initial metadata will be affected by the
/// \a ServerContext associated with this call.
void SendInitialMetadata ( ) override {
GPR_CODEGEN_ASSERT ( ! ctx_ - > sent_initial_metadata_ ) ;
internal : : CallOpSet < internal : : CallOpSendInitialMetadata > ops ;
CallOpSet < CallOpSendInitialMetadata > ops ;
ops . SendInitialMetadata ( ctx_ - > initial_metadata_ ,
ctx_ - > initial_metadata_flags ( ) ) ;
if ( ctx_ - > compression_level_set ( ) ) {
@ -595,29 +537,30 @@ class ServerReader final : public ServerReaderInterface<R> {
}
bool Read ( R * msg ) override {
internal : : CallOpSet < internal : : CallOpRecvMessage < R > > ops ;
CallOpSet < CallOpRecvMessage < R > > ops ;
ops . RecvMessage ( msg ) ;
call_ - > PerformOps ( & ops ) ;
return call_ - > cq ( ) - > Pluck ( & ops ) & & ops . got_message ;
}
private :
internal : : Call * const call_ ;
Call * const call_ ;
ServerContext * const ctx_ ;
template < class ServiceType , class RequestType , class ResponseType >
friend class internal : : ClientStreamingHandler ;
ServerReader ( internal : : Call * call , ServerContext * ctx )
: call_ ( call ) , ctx_ ( ctx ) { }
} ;
/// Server-side interface for streaming writes of message of type \a W.
template < class W >
class ServerWriterInterface : public ServerStreamingInterface ,
public WriterInterface < W > { } ;
/// Synchronous (blocking) server-side API for doing for doing a
/// server-streaming RPCs, where the outgoing message stream coming from the
/// server has messages of type \a W.
template < class W >
class ServerWriter final : public ServerWriterInterface < W > {
public :
ServerWriter ( Call * call , ServerContext * ctx ) : call_ ( call ) , ctx_ ( ctx ) { }
/// See the \a ServerStreamingInterface.SendInitialMetadata method
/// for semantics.
/// Note that initial metadata will be affected by the
@ -625,7 +568,7 @@ class ServerWriter final : public ServerWriterInterface<W> {
void SendInitialMetadata ( ) override {
GPR_CODEGEN_ASSERT ( ! ctx_ - > sent_initial_metadata_ ) ;
internal : : CallOpSet < internal : : CallOpSendInitialMetadata > ops ;
CallOpSet < CallOpSendInitialMetadata > ops ;
ops . SendInitialMetadata ( ctx_ - > initial_metadata_ ,
ctx_ - > initial_metadata_flags ( ) ) ;
if ( ctx_ - > compression_level_set ( ) ) {
@ -641,12 +584,11 @@ class ServerWriter final : public ServerWriterInterface<W> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the
/// \a ClientContext associated with this call to fill in values).
using internal : : WriterInterface < W > : : Write ;
using WriterInterface < W > : : Write ;
bool Write ( const W & msg , WriteOptions options ) override {
if ( options . is_last_message ( ) ) {
options . set_buffer_hint ( ) ;
}
if ( ! ctx_ - > pending_ops_ . SendMessage ( msg , options ) . ok ( ) ) {
return false ;
}
@ -671,16 +613,16 @@ class ServerWriter final : public ServerWriterInterface<W> {
}
private :
internal : : Call * const call_ ;
Call * const call_ ;
ServerContext * const ctx_ ;
template < class ServiceType , class RequestType , class ResponseType >
friend class internal : : ServerStreamingHandler ;
ServerWriter ( internal : : Call * call , ServerContext * ctx )
: call_ ( call ) , ctx_ ( ctx ) { }
} ;
/// Server-side interface for bi-directional streaming.
template < class W , class R >
class ServerReaderWriterInterface : public ServerStreamingInterface ,
public WriterInterface < W > ,
public ReaderInterface < R > { } ;
/// Actual implementation of bi-directional streaming
namespace internal {
template < class W , class R >
@ -746,7 +688,6 @@ class ServerReaderWriterBody final {
Call * const call_ ;
ServerContext * const ctx_ ;
} ;
} // namespace internal
/// Synchronous (blocking) server-side API for a bidirectional
@ -756,6 +697,8 @@ class ServerReaderWriterBody final {
template < class W , class R >
class ServerReaderWriter final : public ServerReaderWriterInterface < W , R > {
public :
ServerReaderWriter ( Call * call , ServerContext * ctx ) : body_ ( call , ctx ) { }
/// See the \a ServerStreamingInterface.SendInitialMetadata method
/// for semantics. Note that initial metadata will be affected by the
/// \a ServerContext associated with this call.
@ -772,18 +715,13 @@ class ServerReaderWriter final : public ServerReaderWriterInterface<W, R> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the \a
/// ServerContext associated with this call).
using internal : : WriterInterface < W > : : Write ;
using WriterInterface < W > : : Write ;
bool Write ( const W & msg , WriteOptions options ) override {
return body_ . Write ( msg , options ) ;
}
private :
internal : : ServerReaderWriterBody < W , R > body_ ;
friend class internal : : TemplatedBidiStreamingHandler < ServerReaderWriter < W , R > ,
false > ;
ServerReaderWriter ( internal : : Call * call , ServerContext * ctx )
: body_ ( call , ctx ) { }
} ;
/// A class to represent a flow-controlled unary call. This is something
@ -798,6 +736,9 @@ template <class RequestType, class ResponseType>
class ServerUnaryStreamer final
: public ServerReaderWriterInterface < ResponseType , RequestType > {
public :
ServerUnaryStreamer ( Call * call , ServerContext * ctx )
: body_ ( call , ctx ) , read_done_ ( false ) , write_done_ ( false ) { }
/// Block to send initial metadata to client.
/// Implicit input parameter:
/// - the \a ServerContext associated with this call will be used for
@ -834,7 +775,7 @@ class ServerUnaryStreamer final
/// \param options The WriteOptions affecting the write operation.
///
/// \return \a true on success, \a false when the stream has been closed.
using internal : : WriterInterface < ResponseType > : : Write ;
using WriterInterface < ResponseType > : : Write ;
bool Write ( const ResponseType & response , WriteOptions options ) override {
if ( write_done_ | | ! read_done_ ) {
return false ;
@ -847,11 +788,6 @@ class ServerUnaryStreamer final
internal : : ServerReaderWriterBody < ResponseType , RequestType > body_ ;
bool read_done_ ;
bool write_done_ ;
friend class internal : : TemplatedBidiStreamingHandler <
ServerUnaryStreamer < RequestType , ResponseType > , true > ;
ServerUnaryStreamer ( internal : : Call * call , ServerContext * ctx )
: body_ ( call , ctx ) , read_done_ ( false ) , write_done_ ( false ) { }
} ;
/// A class to represent a flow-controlled server-side streaming call.
@ -863,6 +799,9 @@ template <class RequestType, class ResponseType>
class ServerSplitStreamer final
: public ServerReaderWriterInterface < ResponseType , RequestType > {
public :
ServerSplitStreamer ( Call * call , ServerContext * ctx )
: body_ ( call , ctx ) , read_done_ ( false ) { }
/// Block to send initial metadata to client.
/// Implicit input parameter:
/// - the \a ServerContext associated with this call will be used for
@ -899,7 +838,7 @@ class ServerSplitStreamer final
/// \param options The WriteOptions affecting the write operation.
///
/// \return \a true on success, \a false when the stream has been closed.
using internal : : WriterInterface < ResponseType > : : Write ;
using WriterInterface < ResponseType > : : Write ;
bool Write ( const ResponseType & response , WriteOptions options ) override {
return read_done_ & & body_ . Write ( response , options ) ;
}
@ -907,11 +846,6 @@ class ServerSplitStreamer final
private :
internal : : ServerReaderWriterBody < ResponseType , RequestType > body_ ;
bool read_done_ ;
friend class internal : : TemplatedBidiStreamingHandler <
ServerSplitStreamer < RequestType , ResponseType > , false > ;
ServerSplitStreamer ( internal : : Call * call , ServerContext * ctx )
: body_ ( call , ctx ) , read_done_ ( false ) { }
} ;
} // namespace grpc
ncteisen
7 years ago