Pass end_of-stream flag to request and response headers. (proxy-wasm#24)

This propagates the end_of_stream flag to the plugin for both
of the header callbacks.

Signed-off-by: Gregory Brail <[email protected]>

Author: gbrail

WORKSPACE

@@ -5,8 +5,8 @@ load("@bazel_tools//tools/build_defs/repo:git.bzl", "git_repository")
 
 git_repository(
     name = "proxy_wasm_cpp_sdk",
+    commit = "b273b07ae0cfa9cc76dfe38236b163e9e3a2ab49",
     remote = "https://github.com/proxy-wasm/proxy-wasm-cpp-sdk",
-    commit = "96927d814b3ec14893b56793e122125e095654c7",
 )
 
 http_archive(
@@ -20,19 +20,21 @@ http_archive(
 # required by com_google_protobuf
 http_archive(
     name = "bazel_skylib",
+    sha256 = "97e70364e9249702246c0e9444bccdc4b847bed1eb03c5a3ece4f83dfe6abc44",
     urls = [
         "https://mirror.bazel.build/github.com/bazelbuild/bazel-skylib/releases/download/1.0.2/bazel-skylib-1.0.2.tar.gz",
         "https://github.com/bazelbuild/bazel-skylib/releases/download/1.0.2/bazel-skylib-1.0.2.tar.gz",
-        ],
-    sha256 = "97e70364e9249702246c0e9444bccdc4b847bed1eb03c5a3ece4f83dfe6abc44",
-    )
+    ],
+)
+
 load("@bazel_skylib//:workspace.bzl", "bazel_skylib_workspace")
+
 bazel_skylib_workspace()
 
 git_repository(
     name = "com_google_protobuf",
-    remote = "https://github.com/protocolbuffers/protobuf",
     commit = "655310ca192a6e3a050e0ca0b7084a2968072260",
+    remote = "https://github.com/protocolbuffers/protobuf",
 )
 
 http_archive(

include/proxy-wasm/null_plugin.h

@@ -17,11 +17,10 @@
 
 #include <memory>
 
+#include "google/protobuf/message.h"
 #include "include/proxy-wasm/null_vm_plugin.h"
 #include "include/proxy-wasm/wasm.h"
 
-#include "google/protobuf/message.h"
-
 namespace proxy_wasm {
 namespace null_plugin {
 #include "proxy_wasm_enums.h"
@@ -84,12 +83,12 @@ class NullPlugin : public NullVmPlugin {
   void onDownstreamConnectionClose(uint64_t context_id, uint64_t close_type);
   void onUpstreamConnectionClose(uint64_t context_id, uint64_t close_type);
 
-  uint64_t onRequestHeaders(uint64_t context_id, uint64_t headers);
+  uint64_t onRequestHeaders(uint64_t context_id, uint64_t headers, uint64_t end_of_stream);
   uint64_t onRequestBody(uint64_t context_id, uint64_t body_buffer_length, uint64_t end_of_stream);
   uint64_t onRequestTrailers(uint64_t context_id, uint64_t trailers);
   uint64_t onRequestMetadata(uint64_t context_id, uint64_t elements);
 
-  uint64_t onResponseHeaders(uint64_t context_id, uint64_t headers);
+  uint64_t onResponseHeaders(uint64_t context_id, uint64_t headers, uint64_t end_of_stream);
   uint64_t onResponseBody(uint64_t context_id, uint64_t body_buffer_length, uint64_t end_of_stream);
   uint64_t onResponseTrailers(uint64_t context_id, uint64_t trailers);
   uint64_t onResponseMetadata(uint64_t context_id, uint64_t elements);

include/proxy-wasm/wasm.h

@@ -20,12 +20,11 @@
 #include <atomic>
 #include <deque>
 #include <map>
+#include <memory>
 #include <unordered_map>
 #include <unordered_set>
-#include <memory>
 
 #include "include/proxy-wasm/compat.h"
-
 #include "include/proxy-wasm/context.h"
 #include "include/proxy-wasm/exports.h"
 #include "include/proxy-wasm/wasm_vm.h"
@@ -206,12 +205,12 @@ class WasmBase : public std::enable_shared_from_this<WasmBase> {
   WasmCallVoid<2> on_downstream_connection_close_;
   WasmCallVoid<2> on_upstream_connection_close_;
 
-  WasmCallWord<2> on_request_headers_;
+  WasmCallWord<3> on_request_headers_;
   WasmCallWord<3> on_request_body_;
   WasmCallWord<2> on_request_trailers_;
   WasmCallWord<2> on_request_metadata_;
 
-  WasmCallWord<2> on_response_headers_;
+  WasmCallWord<3> on_response_headers_;
   WasmCallWord<3> on_response_body_;
   WasmCallWord<2> on_response_trailers_;
   WasmCallWord<2> on_response_metadata_;

src/context.cc

@@ -387,14 +387,15 @@ void ContextBase::onUpstreamConnectionClose(CloseType close_type) {
 // Empty headers/trailers have zero size.
 template <typename P> static uint32_t headerSize(const P &p) { return p ? p->size() : 0; }
 
-FilterHeadersStatus ContextBase::onRequestHeaders(uint32_t headers, bool) {
+FilterHeadersStatus ContextBase::onRequestHeaders(uint32_t headers, bool end_of_stream) {
   DeferAfterCallActions actions(this);
   onCreate(root_context_id_);
   in_vm_context_created_ = true;
   if (!wasm_->on_request_headers_) {
     return FilterHeadersStatus::Continue;
   }
-  auto result = wasm_->on_request_headers_(this, id_, headers).u64_;
+  auto result =
+      wasm_->on_request_headers_(this, id_, headers, static_cast<uint32_t>(end_of_stream)).u64_;
   if (result > static_cast<uint64_t>(FilterHeadersStatus::StopAllIterationAndWatermark))
     return FilterHeadersStatus::StopAllIterationAndWatermark;
   return static_cast<FilterHeadersStatus>(result);
@@ -436,7 +437,7 @@ FilterMetadataStatus ContextBase::onRequestMetadata(uint32_t elements) {
   return FilterMetadataStatus::Continue; // This is currently the only return code.
 }
 
-FilterHeadersStatus ContextBase::onResponseHeaders(uint32_t headers, bool) {
+FilterHeadersStatus ContextBase::onResponseHeaders(uint32_t headers, bool end_of_stream) {
   DeferAfterCallActions actions(this);
   if (!in_vm_context_created_) {
     // If the request is invalid then onRequestHeaders() will not be called and neither will
@@ -449,7 +450,8 @@ FilterHeadersStatus ContextBase::onResponseHeaders(uint32_t headers, bool) {
   if (!wasm_->on_response_headers_) {
     return FilterHeadersStatus::Continue;
   }
-  auto result = wasm_->on_response_headers_(this, id_, headers).u64_;
+  auto result =
+      wasm_->on_response_headers_(this, id_, headers, static_cast<uint32_t>(end_of_stream)).u64_;
   if (result > static_cast<uint64_t>(FilterHeadersStatus::StopAllIterationAndWatermark))
     return FilterHeadersStatus::StopAllIterationAndWatermark;
   return static_cast<FilterHeadersStatus>(result);

src/null/null_plugin.cc

@@ -13,8 +13,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "include/proxy-wasm/null_plugin.h"
-
 #include <stdint.h>
 #include <stdio.h>
 
@@ -25,6 +23,7 @@
 #include <utility>
 #include <vector>
 
+#include "include/proxy-wasm/null_plugin.h"
 #include "include/proxy-wasm/null_vm.h"
 #include "include/proxy-wasm/wasm.h"
 
@@ -175,11 +174,6 @@ void NullPlugin::getFunction(string_view function_name, WasmCallWord<2> *f) {
       SaveRestoreContext saved_context(context);
       return Word(plugin->validateConfiguration(context_id, configuration_size));
     };
-  } else if (function_name == "proxy_on_request_headers") {
-    *f = [plugin](ContextBase *context, Word context_id, Word headers) -> Word {
-      SaveRestoreContext saved_context(context);
-      return Word(plugin->onRequestHeaders(context_id, headers));
-    };
   } else if (function_name == "proxy_on_request_trailers") {
     *f = [plugin](ContextBase *context, Word context_id, Word trailers) -> Word {
       SaveRestoreContext saved_context(context);
@@ -190,11 +184,6 @@ void NullPlugin::getFunction(string_view function_name, WasmCallWord<2> *f) {
       SaveRestoreContext saved_context(context);
       return Word(plugin->onRequestMetadata(context_id, elements));
     };
-  } else if (function_name == "proxy_on_response_headers") {
-    *f = [plugin](ContextBase *context, Word context_id, Word headers) -> Word {
-      SaveRestoreContext saved_context(context);
-      return Word(plugin->onResponseHeaders(context_id, headers));
-    };
   } else if (function_name == "proxy_on_response_trailers") {
     *f = [plugin](ContextBase *context, Word context_id, Word trailers) -> Word {
       SaveRestoreContext saved_context(context);
@@ -225,12 +214,22 @@ void NullPlugin::getFunction(string_view function_name, WasmCallWord<3> *f) {
       SaveRestoreContext saved_context(context);
       return Word(plugin->onUpstreamData(context_id, body_buffer_length, end_of_stream));
     };
+  } else if (function_name == "proxy_on_request_headers") {
+    *f = [plugin](ContextBase *context, Word context_id, Word headers, Word end_of_stream) -> Word {
+      SaveRestoreContext saved_context(context);
+      return Word(plugin->onRequestHeaders(context_id, headers, end_of_stream));
+    };
   } else if (function_name == "proxy_on_request_body") {
     *f = [plugin](ContextBase *context, Word context_id, Word body_buffer_length,
                   Word end_of_stream) -> Word {
       SaveRestoreContext saved_context(context);
       return Word(plugin->onRequestBody(context_id, body_buffer_length, end_of_stream));
     };
+  } else if (function_name == "proxy_on_response_headers") {
+    *f = [plugin](ContextBase *context, Word context_id, Word headers, Word end_of_stream) -> Word {
+      SaveRestoreContext saved_context(context);
+      return Word(plugin->onResponseHeaders(context_id, headers, end_of_stream));
+    };
   } else if (function_name == "proxy_on_response_body") {
     *f = [plugin](ContextBase *context, Word context_id, Word body_buffer_length,
                   Word end_of_stream) -> Word {
@@ -384,8 +383,10 @@ void NullPlugin::onUpstreamConnectionClose(uint64_t context_id, uint64_t close_t
   getContext(context_id)->onUpstreamConnectionClose(static_cast<PeerType>(close_type));
 }
 
-uint64_t NullPlugin::onRequestHeaders(uint64_t context_id, uint64_t headers) {
-  return static_cast<uint64_t>(getContext(context_id)->onRequestHeaders(headers));
+uint64_t NullPlugin::onRequestHeaders(uint64_t context_id, uint64_t headers,
+                                      uint64_t end_of_stream) {
+  return static_cast<uint64_t>(
+      getContext(context_id)->onRequestHeaders(headers, end_of_stream != 0));
 }
 
 uint64_t NullPlugin::onRequestBody(uint64_t context_id, uint64_t body_buffer_length,
@@ -403,8 +404,10 @@ uint64_t NullPlugin::onRequestMetadata(uint64_t context_id, uint64_t elements) {
   return static_cast<uint64_t>(getContext(context_id)->onRequestMetadata(elements));
 }
 
-uint64_t NullPlugin::onResponseHeaders(uint64_t context_id, uint64_t headers) {
-  return static_cast<uint64_t>(getContext(context_id)->onResponseHeaders(headers));
+uint64_t NullPlugin::onResponseHeaders(uint64_t context_id, uint64_t headers,
+                                       uint64_t end_of_stream) {
+  return static_cast<uint64_t>(
+      getContext(context_id)->onResponseHeaders(headers, end_of_stream != 0));
 }
 
 uint64_t NullPlugin::onResponseBody(uint64_t context_id, uint64_t body_buffer_length,