Add support for closing of connections in the ABI/SDK. (proxy-wasm#26)

* Add support for closing of connections in the ABI/SDK.

Signed-off-by: John Plevyak <jplevyak@gmail.com>

Author: jplevyak

WORKSPACE

@@ -5,7 +5,7 @@ load("@bazel_tools//tools/build_defs/repo:git.bzl", "git_repository")
 
 git_repository(
     name = "proxy_wasm_cpp_sdk",
-    commit = "f750d1f5da6a2f20cc55da75dd9772b7ba1650ca",
+    commit = "f44562520bca7bfeee77d6284a96d2900f2f13ac",
     remote = "https://github.com/proxy-wasm/proxy-wasm-cpp-sdk",
 )
 

include/proxy-wasm/context.h

@@ -195,7 +195,7 @@ class ContextBase : public RootInterface,
     unimplemented();
     return nullptr;
   }
-  bool endOfStream(StreamType /* type */) override {
+  bool endOfStream(WasmStreamType /* type */) override {
     unimplemented();
     return true;
   }
@@ -254,7 +254,8 @@ class ContextBase : public RootInterface,
   }
 
   // Continue
-  WasmResult continueStream(StreamType /* stream_type */) override { return unimplemented(); }
+  WasmResult continueStream(WasmStreamType /* stream_type */) override { return unimplemented(); }
+  WasmResult closeStream(WasmStreamType /* stream_type */) override { return unimplemented(); }
   WasmResult sendLocalResponse(uint32_t /* response_code */, string_view /* body_text */,
                                Pairs /* additional_headers */, GrpcStatusCode /* grpc_status */,
                                string_view /* details */) override {

include/proxy-wasm/context_interface.h

@@ -40,34 +40,13 @@ using GrpcStatusCode = uint32_t;
 using SharedQueueDequeueToken = uint32_t;
 using SharedQueueEnqueueToken = uint32_t;
 
-// TODO: move to SDK.
-enum StreamType {
-  Request,
-  Response,
-  Downstream,
-  Upstream,
-};
-
 // TODO: update SDK and use this.
 enum class ProxyAction : uint32_t {
   Illegal = 0,
   Continue = 1,
   Pause = 2,
 };
 
-/*
- * The source of the close event.
- * Unknown is when the source is not known.
- * Local is when the close was initiated by the proxy.
- * Remote is when the close was received from the peer.
- */
-// TODO: update SDK.
-enum class CloseType : uint32_t {
-  Unknown = 0,
-  Local = 1,
-  Remote = 2,
-};
-
 struct PluginBase;
 class WasmBase;
 
@@ -331,8 +310,11 @@ struct NetworkInterface {
  **/
 struct StreamInterface {
   virtual ~StreamInterface() = default;
-  // Continue processing a request e.g. after returning ProxyAction::Pause.
-  virtual WasmResult continueStream(StreamType type) = 0;
+  // Continue processing e.g. after returning ProxyAction::Pause.
+  virtual WasmResult continueStream(WasmStreamType type) = 0;
+
+  // Close a stream.
+  virtual WasmResult closeStream(WasmStreamType type) = 0;
 
   /**
    * Provides a BufferInterface to be used to return buffered data to the VM.
@@ -345,7 +327,7 @@ struct StreamInterface {
    * when a connection is closed (or half-closed) either locally or remotely.
    * @param stream_type is the flow
    */
-  virtual bool endOfStream(StreamType type) = 0;
+  virtual bool endOfStream(WasmStreamType type) = 0;
 };
 
 // Header/Trailer/Metadata Maps

include/proxy-wasm/exports.h

@@ -29,8 +29,8 @@ Word log(void *raw_context, Word level, Word address, Word size);
 Word get_property(void *raw_context, Word path_ptr, Word path_size, Word value_ptr_ptr,
                   Word value_size_ptr);
 Word set_property(void *raw_context, Word key_ptr, Word key_size, Word value_ptr, Word value_size);
-Word continue_request(void *raw_context);
-Word continue_response(void *raw_context);
+Word continue_stream(void *raw_context, Word stream_type);
+Word close_stream(void *raw_context, Word stream_type);
 Word send_local_response(void *raw_context, Word response_code, Word response_code_details_ptr,
                          Word response_code_details_size, Word body_ptr, Word body_size,
                          Word additional_response_header_pairs_ptr,

include/proxy-wasm/wasm_api_impl.h

@@ -65,11 +65,11 @@ inline WasmResult proxy_set_property(const char *key_ptr, size_t key_size, const
 }
 
 // Continue
-inline WasmResult proxy_continue_request() {
-  return wordToWasmResult(exports::continue_request(current_context_));
+inline WasmResult proxy_continue_stream(WasmStreamType stream_type) {
+  return wordToWasmResult(exports::continue_stream(current_context_, WS(stream_type)));
 }
-inline WasmResult proxy_continue_response() {
-  return wordToWasmResult(exports::continue_response(current_context_));
+inline WasmResult proxy_close_stream(WasmStreamType stream_type) {
+  return wordToWasmResult(exports::close_stream(current_context_, WS(stream_type)));
 }
 inline WasmResult
 proxy_send_local_response(uint32_t response_code, const char *response_code_details_ptr,

src/exports.cc

@@ -164,16 +164,20 @@ Word get_status(void *raw_context, Word code_ptr, Word value_ptr_ptr, Word value
 // HTTP
 
 // Continue/Reply/Route
-Word continue_request(void *raw_context) {
+Word continue_stream(void *raw_context, Word type) {
   auto context = WASM_CONTEXT(raw_context);
-  context->continueStream(StreamType::Request);
-  return WasmResult::Ok;
+  if (type > static_cast<uint64_t>(WasmStreamType::MAX)) {
+    return WasmResult::BadArgument;
+  }
+  return context->continueStream(static_cast<WasmStreamType>(type.u64_));
 }
 
-Word continue_response(void *raw_context) {
+Word close_stream(void *raw_context, Word type) {
   auto context = WASM_CONTEXT(raw_context);
-  context->continueStream(StreamType::Response);
-  return WasmResult::Ok;
+  if (type > static_cast<uint64_t>(WasmStreamType::MAX)) {
+    return WasmResult::BadArgument;
+  }
+  return context->closeStream(static_cast<WasmStreamType>(type.u64_));
 }
 
 Word send_local_response(void *raw_context, Word response_code, Word response_code_details_ptr,

src/null/null_plugin.cc

@@ -376,11 +376,11 @@ uint64_t NullPlugin::onUpstreamData(uint64_t context_id, uint64_t data_length,
 }
 
 void NullPlugin::onDownstreamConnectionClose(uint64_t context_id, uint64_t close_type) {
-  getContext(context_id)->onDownstreamConnectionClose(static_cast<PeerType>(close_type));
+  getContext(context_id)->onDownstreamConnectionClose(static_cast<CloseType>(close_type));
 }
 
 void NullPlugin::onUpstreamConnectionClose(uint64_t context_id, uint64_t close_type) {
-  getContext(context_id)->onUpstreamConnectionClose(static_cast<PeerType>(close_type));
+  getContext(context_id)->onUpstreamConnectionClose(static_cast<CloseType>(close_type));
 }
 
 uint64_t NullPlugin::onRequestHeaders(uint64_t context_id, uint64_t headers,

src/wasm.cc

@@ -146,8 +146,8 @@ void WasmBase::registerCallbacks() {
   _REGISTER_PROXY(set_property);
   _REGISTER_PROXY(get_property);
 
-  _REGISTER_PROXY(continue_request);
-  _REGISTER_PROXY(continue_response);
+  _REGISTER_PROXY(continue_stream);
+  _REGISTER_PROXY(close_stream);
   _REGISTER_PROXY(send_local_response);
 
   _REGISTER_PROXY(get_shared_data);