Initial attempt at a C++ wrapper for the C grpc_transport_op and

grpc_transport_stream_op structs.
pull/6915/head
Mark D. Roth 9 years ago
parent 34d07d6f2a
commit 07cd9c9e06
  1. 167
      include/grpc++/channel_filter.h
  2. 8
      src/core/lib/transport/metadata.h
  3. 8
      src/core/lib/transport/metadata_batch.h
  4. 23
      src/cpp/common/channel_filter.cc
  5. 6
      test/cpp/end2end/filter_end2end_test.cc
  6. 2
      third_party/protobuf

@ -35,13 +35,16 @@
#define GRPCXX_CHANNEL_FILTER_H
#include <grpc/grpc.h>
#include <grpc/census.h>
#include <grpc++/impl/codegen/config.h>
#include <functional>
#include <vector>
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/security/context/security_context.h"
#include "src/core/lib/surface/channel_init.h"
#include "src/core/lib/transport/metadata_batch.h"
//
// An interface to define filters.
@ -54,16 +57,164 @@
namespace grpc {
// A C++ wrapper for the grpc_metadata_batch struct.
class MetadataBatch {
public:
explicit MetadataBatch(grpc_metadata_batch* batch) : batch_(batch) {}
grpc_metadata_batch* batch() const { return batch_; }
// Adds metadata and returns the newly allocated storage.
// The caller takes ownership of the result, which must exist for the
// lifetime of the gRPC call.
grpc_linked_mdelem* AddMetadata(const string& key, const string& value);
class const_iterator : public std::iterator<std::bidirectional_iterator_tag,
const grpc_mdelem> {
public:
const grpc_mdelem& operator*() const { return *elem_->md; }
const grpc_mdelem* operator->() const { return elem_->md; }
const_iterator& operator++() {
elem_ = elem_->next;
return *this;
}
const_iterator operator++(int) {
const_iterator tmp(*this);
operator++();
return tmp;
}
const_iterator& operator--() {
elem_ = elem_->prev;
return *this;
}
const_iterator operator--(int) {
const_iterator tmp(*this);
operator--();
return tmp;
}
bool operator==(const const_iterator& other) const {
return elem_ == other.elem_;
}
bool operator!=(const const_iterator& other) const {
return elem_ != other.elem_;
}
private:
friend class MetadataBatch;
explicit const_iterator(grpc_linked_mdelem* elem) : elem_(elem) {}
grpc_linked_mdelem* elem_;
};
const_iterator begin() const { return const_iterator(batch_->list.head); }
const_iterator end() const { return const_iterator(nullptr); }
private:
grpc_metadata_batch* batch_;
};
// A C++ wrapper for the grpc_transport_op struct.
class TransportOp {
public:
explicit TransportOp(grpc_transport_op* op) : op_(op) {}
grpc_transport_op* op() const { return op_; }
bool disconnect() const { return op_->disconnect; }
bool send_goaway() const { return op_->send_goaway; }
// TODO(roth): Add methods for additional fields as needed.
private:
grpc_transport_op* op_; // Do not own.
};
// A C++ wrapper for the grpc_transport_stream_op struct.
class TransportStreamOp {
public:
explicit TransportStreamOp(grpc_transport_stream_op* op)
: op_(op),
send_initial_metadata_(op->send_initial_metadata),
send_trailing_metadata_(op->send_trailing_metadata),
recv_initial_metadata_(op->recv_initial_metadata),
recv_trailing_metadata_(op->recv_trailing_metadata) {}
grpc_transport_stream_op* op() const { return op_; }
grpc_closure* on_complete() const { return op_->on_complete; }
void set_on_complete(grpc_closure* closure) {
op_->on_complete = closure;
}
MetadataBatch* send_initial_metadata() {
return op_->send_initial_metadata == nullptr
? nullptr : &send_initial_metadata_;
}
MetadataBatch* send_trailing_metadata() {
return op_->send_trailing_metadata == nullptr
? nullptr : &send_trailing_metadata_;
}
MetadataBatch* recv_initial_metadata() {
return op_->recv_initial_metadata == nullptr
? nullptr : &recv_initial_metadata_;
}
MetadataBatch* recv_trailing_metadata() {
return op_->recv_trailing_metadata == nullptr
? nullptr : &recv_trailing_metadata_;
}
uint32_t* send_initial_metadata_flags() const {
return &op_->send_initial_metadata_flags;
}
grpc_closure* recv_initial_metadata_ready() const {
return op_->recv_initial_metadata_ready;
}
void set_recv_initial_metadata_ready(grpc_closure* closure) {
op_->recv_initial_metadata_ready = closure;
}
grpc_byte_stream* send_message() const { return op_->send_message; }
void set_send_message(grpc_byte_stream* send_message) {
op_->send_message = send_message;
}
// To be called only on clients and servers, respectively.
grpc_client_security_context* client_security_context() const {
return (grpc_client_security_context*)op_->context[
GRPC_CONTEXT_SECURITY].value;
}
grpc_server_security_context* server_security_context() const {
return (grpc_server_security_context*)op_->context[
GRPC_CONTEXT_SECURITY].value;
}
census_context* get_census_context() const {
return (census_context*)op_->context[GRPC_CONTEXT_TRACING].value;
}
private:
grpc_transport_stream_op* op_; // Do not own.
MetadataBatch send_initial_metadata_;
MetadataBatch send_trailing_metadata_;
MetadataBatch recv_initial_metadata_;
MetadataBatch recv_trailing_metadata_;
};
// Represents channel data.
class ChannelData {
public:
virtual ~ChannelData() {}
const char* peer() const { return peer_; }
// FIXME: find a way to avoid passing elem into these methods
// (same for CallData below)
virtual void StartTransportOp(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_transport_op *op);
const char* peer() const { return peer_; }
TransportOp *op);
protected:
ChannelData(const grpc_channel_args &args, const char *peer) : peer_(peer) {}
@ -79,7 +230,7 @@ class CallData {
virtual void StartTransportStreamOp(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op);
TransportStreamOp *op);
virtual void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
@ -88,6 +239,8 @@ class CallData {
virtual char *GetPeer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem);
protected:
// FIXME: once PR #7024 has been merged, update this API to provide a
// way to return an error from call initialization
explicit CallData(const ChannelData &) {}
};
@ -119,7 +272,8 @@ class ChannelFilter GRPC_FINAL {
grpc_channel_element *elem,
grpc_transport_op *op) {
ChannelDataType *channel_data = (ChannelDataType *)elem->channel_data;
channel_data->StartTransportOp(exec_ctx, elem, op);
TransportOp op_wrapper(op);
channel_data->StartTransportOp(exec_ctx, elem, &op_wrapper);
}
static const size_t call_data_size = sizeof(CallDataType);
@ -143,7 +297,8 @@ class ChannelFilter GRPC_FINAL {
grpc_call_element *elem,
grpc_transport_stream_op *op) {
CallDataType *call_data = (CallDataType *)elem->call_data;
call_data->StartTransportStreamOp(exec_ctx, elem, op);
TransportStreamOp op_wrapper(op);
call_data->StartTransportStreamOp(exec_ctx, elem, &op_wrapper);
}
static void SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,

@ -37,6 +37,10 @@
#include <grpc/support/slice.h>
#include <grpc/support/useful.h>
#ifdef __cplusplus
extern "C" {
#endif
/* This file provides a mechanism for tracking metadata through the grpc stack.
It's not intended for consumption outside of the library.
@ -164,4 +168,8 @@ void grpc_mdctx_global_shutdown(void);
extern gpr_slice (*grpc_chttp2_base64_encode_and_huffman_compress)(
gpr_slice input);
#ifdef __cplusplus
}
#endif
#endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_H */

@ -42,6 +42,10 @@
#include <grpc/support/time.h>
#include "src/core/lib/transport/metadata.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct grpc_linked_mdelem {
grpc_mdelem *md;
struct grpc_linked_mdelem *next;
@ -127,4 +131,8 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch *comd);
} while (0)
#endif
#ifdef __cplusplus
}
#endif
#endif /* GRPC_CORE_LIB_TRANSPORT_METADATA_BATCH_H */

@ -33,18 +33,33 @@
#include <grpc++/channel_filter.h>
#include <string.h>
#include "src/core/lib/channel/channel_stack.h"
namespace grpc {
//
// MetadataBatch
//
grpc_linked_mdelem* MetadataBatch::AddMetadata(
const string& key, const string& value) {
grpc_linked_mdelem *storage = new grpc_linked_mdelem;
memset(storage, 0, sizeof(grpc_linked_mdelem));
storage->md = grpc_mdelem_from_strings(key.c_str(), value.c_str());
grpc_metadata_batch_link_head(batch_, storage);
return storage;
}
//
// ChannelData
//
void ChannelData::StartTransportOp(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_transport_op *op) {
grpc_channel_next_op(exec_ctx, elem, op);
TransportOp *op) {
grpc_channel_next_op(exec_ctx, elem, op->op());
}
//
@ -53,8 +68,8 @@ void ChannelData::StartTransportOp(grpc_exec_ctx *exec_ctx,
void CallData::StartTransportStreamOp(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op) {
grpc_call_next_op(exec_ctx, elem, op);
TransportStreamOp *op) {
grpc_call_next_op(exec_ctx, elem, op->op());
}
void CallData::SetPollsetOrPollsetSet(grpc_exec_ctx *exec_ctx,

@ -125,11 +125,11 @@ class CallDataImpl : public CallData {
: CallData(channel_data) {}
void StartTransportStreamOp(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
grpc_transport_stream_op* op) GRPC_OVERRIDE {
TransportStreamOp* op) GRPC_OVERRIDE {
// Incrementing the counter could be done from the ctor, but we want
// to test that the individual methods are actually called correctly.
if (op->recv_initial_metadata != nullptr) IncrementCallCounter();
grpc_call_next_op(exec_ctx, elem, op);
if (op->recv_initial_metadata() != nullptr) IncrementCallCounter();
grpc_call_next_op(exec_ctx, elem, op->op());
}
};

@ -1 +1 @@
Subproject commit 3470b6895aa659b7559ed678e029a5338e535f14
Subproject commit d4d13a4349e4e59d67f311185ddcc1890d956d7a
Loading…
Cancel
Save