Merge pull request #2406 from dgquintas/writeoptions

Added WriteOptions for per-message flags and modified CallOps to use it.

include/grpc++/impl/call.h

@@ -60,6 +60,93 @@ void FillMetadataMap(grpc_metadata_array* arr,
 grpc_metadata* FillMetadataArray(
     const std::multimap<grpc::string, grpc::string>& metadata);
 
+/// Per-message write options.
+class WriteOptions {
+ public:
+  WriteOptions() : flags_(0) {}
+  WriteOptions(const WriteOptions& other) : flags_(other.flags_) {}
+
+  /// Clear all flags.
+  inline void Clear() {
+    flags_ = 0;
+  }
+
+  /// Returns raw flags bitset.
+  inline gpr_uint32 flags() const {
+    return flags_;
+  }
+
+  /// Sets flag for the disabling of compression for the next message write.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline WriteOptions& set_no_compression() {
+    SetBit(GRPC_WRITE_NO_COMPRESS);
+    return *this;
+  }
+
+  /// Clears flag for the disabling of compression for the next message write.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline WriteOptions& clear_no_compression() {
+    ClearBit(GRPC_WRITE_NO_COMPRESS);
+    return *this;
+  }
+
+  /// Get value for the flag indicating whether compression for the next
+  /// message write is forcefully disabled.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline bool get_no_compression() const {
+    return GetBit(GRPC_WRITE_NO_COMPRESS);
+  }
+
+  /// Sets flag indicating that the write may be buffered and need not go out on
+  /// the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline WriteOptions& set_buffer_hint() {
+    SetBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  /// Clears flag indicating that the write may be buffered and need not go out
+  /// on the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline WriteOptions& clear_buffer_hint() {
+    ClearBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  /// Get value for the flag indicating that the write may be buffered and need
+  /// not go out on the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline bool get_buffer_hint() const {
+    return GetBit(GRPC_WRITE_BUFFER_HINT);
+  }
+
+  WriteOptions& operator=(const WriteOptions& rhs) {
+    flags_ = rhs.flags_;
+    return *this;
+  }
+
+ private:
+  void SetBit(const gpr_int32 mask) {
+    flags_ |= mask;
+  }
+
+  void ClearBit(const gpr_int32 mask) {
+    flags_ &= ~mask;
+  }
+
+  bool GetBit(const gpr_int32 mask) const {
+    return flags_ & mask;
+  }
+
+  gpr_uint32 flags_;
+};
+
 /// Default argument for CallOpSet. I is unused by the class, but can be
 /// used for generating multiple names for the same thing.
 template <int I>
@@ -104,6 +191,12 @@ class CallOpSendMessage {
  public:
   CallOpSendMessage() : send_buf_(nullptr), own_buf_(false) {}
 
+  /// Send \a message using \a options for the write. The \a options are cleared
+  /// after use.
+  template <class M>
+  Status SendMessage(const M& message,
+                     const WriteOptions& options) GRPC_MUST_USE_RESULT;
+
   template <class M>
   Status SendMessage(const M& message) GRPC_MUST_USE_RESULT;
 
@@ -112,8 +205,10 @@ class CallOpSendMessage {
     if (send_buf_ == nullptr) return;
     grpc_op* op = &ops[(*nops)++];
     op->op = GRPC_OP_SEND_MESSAGE;
-    op->flags = 0;
+    op->flags = write_options_.flags();
     op->data.send_message = send_buf_;
+    // Flags are per-message: clear them after use.
+    write_options_.Clear();
   }
   void FinishOp(bool* status, int max_message_size) {
     if (own_buf_) grpc_byte_buffer_destroy(send_buf_);
@@ -122,14 +217,22 @@ class CallOpSendMessage {
 
  private:
   grpc_byte_buffer* send_buf_;
+  WriteOptions write_options_;
   bool own_buf_;
 };
 
 template <class M>
-Status CallOpSendMessage::SendMessage(const M& message) {
+Status CallOpSendMessage::SendMessage(const M& message,
+                                      const WriteOptions& options) {
+  write_options_ = options;
   return SerializationTraits<M>::Serialize(message, &send_buf_, &own_buf_);
 }
 
+template <class M>
+Status CallOpSendMessage::SendMessage(const M& message) {
+  return SendMessage(message, WriteOptions());
+}
+
 template <class R>
 class CallOpRecvMessage {
  public:

include/grpc++/stream.h

@@ -79,7 +79,11 @@ class 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;
+  virtual bool Write(const W& msg, const WriteOptions& options) = 0;
+
+  inline bool Write(const W& msg) {
+    return Write(msg, WriteOptions());
+  }
 };
 
 template <class R>
@@ -168,9 +172,10 @@ class ClientWriter : public ClientWriterInterface<W> {
     cq_.Pluck(&ops);
   }
 
-  bool Write(const W& msg) GRPC_OVERRIDE {
+  using WriterInterface<W>::Write;
+  bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
     CallOpSet<CallOpSendMessage> ops;
-    if (!ops.SendMessage(msg).ok()) {
+    if (!ops.SendMessage(msg, options).ok()) {
       return false;
     }
     call_.PerformOps(&ops);
@@ -246,9 +251,10 @@ class ClientReaderWriter GRPC_FINAL : public ClientReaderWriterInterface<W, R> {
     return cq_.Pluck(&ops) && ops.got_message;
   }
 
-  bool Write(const W& msg) GRPC_OVERRIDE {
+  using WriterInterface<W>::Write;
+  bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
     CallOpSet<CallOpSendMessage> ops;
-    if (!ops.SendMessage(msg).ok()) return false;
+    if (!ops.SendMessage(msg, options).ok()) return false;
     call_.PerformOps(&ops);
     return cq_.Pluck(&ops);
   }
@@ -317,9 +323,10 @@ class ServerWriter GRPC_FINAL : public WriterInterface<W> {
     call_->cq()->Pluck(&ops);
   }
 
-  bool Write(const W& msg) GRPC_OVERRIDE {
+  using WriterInterface<W>::Write;
+  bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
-    if (!ops.SendMessage(msg).ok()) {
+    if (!ops.SendMessage(msg, options).ok()) {
       return false;
     }
     if (!ctx_->sent_initial_metadata_) {
@@ -359,9 +366,10 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>,
     return call_->cq()->Pluck(&ops) && ops.got_message;
   }
 
-  bool Write(const W& msg) GRPC_OVERRIDE {
+  using WriterInterface<W>::Write;
+  bool Write(const W& msg, const WriteOptions& options) GRPC_OVERRIDE {
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> ops;
-    if (!ops.SendMessage(msg).ok()) {
+    if (!ops.SendMessage(msg, options).ok()) {
       return false;
     }
     if (!ctx_->sent_initial_metadata_) {

test/cpp/end2end/mock_test.cc

@@ -86,7 +86,9 @@ class MockClientReaderWriter<EchoRequest, EchoResponse> GRPC_FINAL
     msg->set_message(last_message_);
     return true;
   }
-  bool Write(const EchoRequest& msg) GRPC_OVERRIDE {
+
+  bool Write(const EchoRequest& msg,
+             const WriteOptions& options) GRPC_OVERRIDE {
     gpr_log(GPR_INFO, "mock recv msg %s", msg.message().c_str());
     last_message_ = msg.message();
     return true;