core: dispatcher test suite and stream completion

library/common/http/dispatcher.cc

@@ -6,41 +6,48 @@
 namespace Envoy {
 namespace Http {
 
+// FIXME general questions:
+// 1. We say at the interface level in envoy that sendHeaders must only be called once and before
+// sendData. But I can't see where that guarantee is kept?
+// 2. What prevents that sendXXX is not called after `end_stream == true` and the stream is
+// local_closed?
+
 Dispatcher::DirectStreamCallbacks::DirectStreamCallbacks(envoy_stream_t stream,
                                                          envoy_observer observer,
                                                          Dispatcher& http_dispatcher)
     : stream_(stream), observer_(observer), http_dispatcher_(http_dispatcher) {}
 
 void Dispatcher::DirectStreamCallbacks::onHeaders(HeaderMapPtr&& headers, bool end_stream) {
-  ENVOY_LOG(debug, "response headers for stream (end_stream={}):\n{}", end_stream, *headers);
-  if (end_stream) {
-    http_dispatcher_.removeStream(stream_);
-  }
+  ENVOY_LOG(debug, "[S{}] response headers for stream (end_stream={}):\n{}", stream_, end_stream,
+            *headers);
   observer_.on_headers_f(Utility::transformHeaders(*headers), end_stream, observer_.context);
+  http_dispatcher_.closeRemote(stream_, end_stream);
 }
 
 void Dispatcher::DirectStreamCallbacks::onData(Buffer::Instance& data, bool end_stream) {
-  ENVOY_LOG(debug, "response data for stream (length={} end_stream={})", data.length(), end_stream);
-  if (end_stream) {
-    http_dispatcher_.removeStream(stream_);
-  }
+  ENVOY_LOG(debug, "[S{}] response data for stream (length={} end_stream={})", stream_,
+            data.length(), end_stream);
   observer_.on_data_f(Envoy::Buffer::Utility::transformData(data), end_stream, observer_.context);
+  http_dispatcher_.closeRemote(stream_, end_stream);
 }
 
 void Dispatcher::DirectStreamCallbacks::onTrailers(HeaderMapPtr&& trailers) {
-  ENVOY_LOG(debug, "response trailers for stream:\n{}", *trailers);
-  http_dispatcher_.removeStream(stream_);
+  ENVOY_LOG(debug, "[S{}] response trailers for stream:\n{}", stream_, *trailers);
   observer_.on_trailers_f(Utility::transformHeaders(*trailers), observer_.context);
+  http_dispatcher_.closeRemote(stream_, true);
 }
 
 void Dispatcher::DirectStreamCallbacks::onReset() {
-  http_dispatcher_.removeStream(stream_);
+  ENVOY_LOG(debug, "[S{}] remote reset stream", stream_);
   observer_.on_error_f({ENVOY_STREAM_RESET, {0, nullptr}}, observer_.context);
+  http_dispatcher_.closeRemote(stream_, true);
 }
 
-Dispatcher::DirectStream::DirectStream(AsyncClient::Stream& underlying_stream,
+Dispatcher::DirectStream::DirectStream(envoy_stream_t stream_id,
+                                       AsyncClient::Stream& underlying_stream,
                                        DirectStreamCallbacksPtr&& callbacks)
-    : underlying_stream_(underlying_stream), callbacks_(std::move(callbacks)) {}
+    : stream_id_(stream_id), underlying_stream_(underlying_stream),
+      callbacks_(std::move(callbacks)) {}
 
 Dispatcher::Dispatcher(Event::Dispatcher& event_dispatcher,
                        Upstream::ClusterManager& cluster_manager)
@@ -62,9 +69,9 @@ envoy_stream_t Dispatcher::startStream(envoy_observer observer) {
       callbacks->onReset();
     } else {
       DirectStreamPtr direct_stream =
-          std::make_unique<DirectStream>(*underlying_stream, std::move(callbacks));
+          std::make_unique<DirectStream>(new_stream_id, *underlying_stream, std::move(callbacks));
       streams_.emplace(new_stream_id, std::move(direct_stream));
-      ENVOY_LOG(debug, "started stream [{}]", new_stream_id);
+      ENVOY_LOG(debug, "[S{}] start stream", new_stream_id);
     }
   });
 
@@ -84,9 +91,10 @@ envoy_status_t Dispatcher::sendHeaders(envoy_stream_t stream_id, envoy_headers h
     // https://github.com/lyft/envoy-mobile/issues/301
     if (direct_stream != nullptr) {
       direct_stream->headers_ = Utility::transformHeaders(headers);
-      ENVOY_LOG(debug, "request headers for stream [{}] (end_stream={}):\n{}", stream_id,
+      ENVOY_LOG(debug, "[S{}] request headers for stream (end_stream={}):\n{}", stream_id,
                 end_stream, *direct_stream->headers_);
       direct_stream->underlying_stream_.sendHeaders(*direct_stream->headers_, end_stream);
+      closeLocal(stream_id, end_stream);
     }
   });
 
@@ -99,8 +107,29 @@ envoy_status_t Dispatcher::sendMetadata(envoy_stream_t, envoy_headers, bool) {
   return ENVOY_FAILURE;
 }
 envoy_status_t Dispatcher::sendTrailers(envoy_stream_t, envoy_headers) { return ENVOY_FAILURE; }
-envoy_status_t Dispatcher::locallyCloseStream(envoy_stream_t) { return ENVOY_FAILURE; }
-envoy_status_t Dispatcher::resetStream(envoy_stream_t) { return ENVOY_FAILURE; }
+
+envoy_status_t Dispatcher::locallyCloseStream(envoy_stream_t stream_id) {
+  event_dispatcher_.post([this, stream_id]() -> void {
+    // FIXME underlying stream does not have a locallyClose function. But we need to inform the
+    // underlying stream of local close if it came through here.
+    closeLocal(stream_id, true);
+    ENVOY_LOG(debug, "[S{}] locally close stream", stream_id);
+  });
+  return ENVOY_SUCCESS;
+}
+
+envoy_status_t Dispatcher::resetStream(envoy_stream_t stream_id) {
+  event_dispatcher_.post([this, stream_id]() -> void {
+    DirectStream* direct_stream = getStream(stream_id);
+    if (direct_stream) {
+      // Resetting the underlying_stream will fire the onReset callback, and thus closeRemote.
+      direct_stream->underlying_stream_.reset();
+      closeLocal(direct_stream->stream_id_, true);
+      ENVOY_LOG(debug, "[S{}] local reset stream", stream_id);
+    }
+  });
+  return ENVOY_SUCCESS;
+}
 
 Dispatcher::DirectStream* Dispatcher::getStream(envoy_stream_t stream_id) {
   ASSERT(event_dispatcher_.isThreadSafe(),
@@ -109,9 +138,45 @@ Dispatcher::DirectStream* Dispatcher::getStream(envoy_stream_t stream_id) {
   return (direct_stream_pair_it != streams_.end()) ? direct_stream_pair_it->second.get() : nullptr;
 }
 
-// TODO: implement. Note: the stream might not be in the map if for example startStream called
-// onReset due to its inability to get an underlying stream.
-envoy_status_t Dispatcher::removeStream(envoy_stream_t) { return ENVOY_FAILURE; }
+void Dispatcher::closeLocal(envoy_stream_t stream_id, bool end_stream) {
+  DirectStream* direct_stream = getStream(stream_id);
+  // FIXME: why would there be no stream?
+  if (direct_stream) {
+    direct_stream->local_closed_ = direct_stream->local_closed_ || end_stream;
+    ENVOY_LOG(debug, "[S{}] local close for stream (local_closed={} remote_closed={})", stream_id,
+              direct_stream->local_closed_, direct_stream->remote_closed_);
+    if (direct_stream->complete()) {
+      cleanup(*direct_stream);
+    }
+  }
+}
+
+void Dispatcher::closeRemote(envoy_stream_t stream_id, bool end_stream) {
+  DirectStream* direct_stream = getStream(stream_id);
+  // FIXME: why would there be no stream?
+  if (direct_stream) {
+    direct_stream->remote_closed_ = direct_stream->remote_closed_ || end_stream;
+    ENVOY_LOG(debug, "[S{}] remote close for stream (local_closed={} remote_closed={})", stream_id,
+              direct_stream->local_closed_, direct_stream->remote_closed_);
+    if (direct_stream->complete()) {
+      cleanup(*direct_stream);
+    }
+  }
+}
+
+void Dispatcher::cleanup(DirectStream& direct_stream) {
+  ASSERT(event_dispatcher_.isThreadSafe(),
+         "stream interaction must be performed on the event_dispatcher_'s thread.");
+  ENVOY_LOG(debug, "[S{}] cleanup for stream", direct_stream.stream_id_);
+  direct_stream.local_closed_ = direct_stream.remote_closed_ = true;
+
+  // FIXME this should probably be deferred so that callbacks are able to fire.
+  // Mmmm actually, is that the case?
+  size_t erased = streams_.erase(direct_stream.stream_id_);
+  if (erased > 0) {
+    ENVOY_LOG(debug, "[S{}] remove stream", direct_stream.stream_id_);
+  }
+}
 
 } // namespace Http
 } // namespace Envoy

library/common/http/dispatcher.h

@@ -32,17 +32,16 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
    * @return envoy_stream_t handle to the stream being created.
    */
   envoy_stream_t startStream(envoy_observer observer);
-  envoy_status_t sendHeaders(envoy_stream_t stream, envoy_headers headers, bool end_stream);
-  envoy_status_t sendData(envoy_stream_t stream, envoy_headers headers, bool end_stream);
-  envoy_status_t sendMetadata(envoy_stream_t stream, envoy_headers headers, bool end_stream);
-  envoy_status_t sendTrailers(envoy_stream_t stream, envoy_headers headers);
-  envoy_status_t locallyCloseStream(envoy_stream_t stream);
+  envoy_status_t sendHeaders(envoy_stream_t stream_id, envoy_headers headers, bool end_stream);
+  envoy_status_t sendData(envoy_stream_t stream_id, envoy_headers headers, bool end_stream);
+  envoy_status_t sendMetadata(envoy_stream_t stream_id, envoy_headers headers, bool end_stream);
+  envoy_status_t sendTrailers(envoy_stream_t stream_id, envoy_headers headers);
+  envoy_status_t locallyCloseStream(envoy_stream_t stream_id);
   // TODO: when implementing this function we have to make sure to prevent races with already
   // scheduled and potentially scheduled callbacks. In order to do so the platform callbacks need to
   // check for atomic state (boolean most likely) that will be updated here to mark the stream as
   // closed.
-  envoy_status_t resetStream(envoy_stream_t stream);
-  envoy_status_t removeStream(envoy_stream_t stream);
+  envoy_status_t resetStream(envoy_stream_t stream_id);
 
 private:
   /**
@@ -77,13 +76,12 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
    * AsyncClient::Stream and in the incoming direction via DirectStreamCallbacks.
    */
   class DirectStream {
-    // TODO: Bookkeeping for this class is insufficient to fully cover all cases necessary to
-    // track the lifecycle of the underlying_stream_. One way or another, we must fix this
-    // to prevent bugs in the future. (Enhanced internal bookkeeping is probably good enough,
-    // but other options include upstream modifications to AsyncClient and friends.
   public:
-    DirectStream(AsyncClient::Stream& underlying_stream, DirectStreamCallbacksPtr&& callbacks);
+    DirectStream(envoy_stream_t stream_id, AsyncClient::Stream& underlying_stream,
+                 DirectStreamCallbacksPtr&& callbacks);
+    bool complete() { return local_closed_ && remote_closed_; }
 
+    const envoy_stream_t stream_id_;
     // Used to issue outgoing HTTP stream operations.
     AsyncClient::Stream& underlying_stream_;
     // Used to receive incoming HTTP stream operations.
@@ -95,13 +93,19 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
     // An implementation option would be to have drainable header maps, or done callbacks.
     std::vector<HeaderMapPtr> metadata_;
     HeaderMapPtr trailers_;
+
+    std::atomic<bool> local_closed_{};
+    std::atomic<bool> remote_closed_{};
   };
 
   using DirectStreamPtr = std::unique_ptr<DirectStream>;
 
   // Everything in the below interface must only be accessed from the event_dispatcher's thread.
   // This allows us to generally avoid synchronization.
   DirectStream* getStream(envoy_stream_t stream_id);
+  void cleanup(DirectStream& stream_id);
+  void closeLocal(envoy_stream_t stream_id, bool end_stream);
+  void closeRemote(envoy_stream_t stream_id, bool end_stream);
 
   std::unordered_map<envoy_stream_t, DirectStreamPtr> streams_;
   std::atomic<envoy_stream_t> current_stream_id_;

library/objective-c/AppDelegate.h

@@ -0,0 +1,8 @@
+#import <Foundation/Foundation.h>
+#import <UIKit/UIKit.h>
+
+@interface AppDelegate : UIResponder <UIApplicationDelegate>
+
+@property (strong, nonatomic) UIWindow *window;
+
+@end

library/objective-c/AppDelegate.mm

@@ -0,0 +1,68 @@
+#import "AppDelegate.h"
+#import <UIKit/UIKit.h>
+#import "ViewController.h"
+#include "library/objective-c/EnvoyEngine.h"
+
+@interface AppDelegate () <NSURLConnectionDelegate>
+@end
+
+@implementation AppDelegate
+
+- (BOOL)application:(UIApplication *)application
+    didFinishLaunchingWithOptions:(NSDictionary *)launchOptions {
+  [NSThread detachNewThreadSelector:@selector(envoyRunner) toTarget:self withObject:nil];
+  [NSThread detachNewThreadSelector:@selector(envoyHarness) toTarget:self withObject:nil];
+
+  UIViewController *controller = [UITableViewController new];
+  self.window = [[UIWindow alloc] initWithFrame:[UIScreen mainScreen].bounds];
+  [self.window setRootViewController:controller];
+  [self.window makeKeyAndVisible];
+  NSLog(@"Finished launching!");
+  return YES;
+}
+
+- (void)envoyRunner {
+  // Read in and store as string
+  NSString *configFile = [[NSBundle mainBundle] pathForResource:@"config" ofType:@"yaml"];
+  NSString *configYaml = [NSString stringWithContentsOfFile:configFile
+                                                   encoding:NSUTF8StringEncoding
+                                                      error:NULL];
+  NSLog(@"Loading config: %@", configYaml);
+
+  [EnvoyEngine runWithConfig:configYaml logLevel:@"debug"];
+}
+
+- (void)envoyHarness {
+  sleep(5);
+  [EnvoyEngine setupEnvoy];
+  sleep(1);
+  while (true) {
+    [EnvoyEngine makeRequest];
+    sleep(1);
+  }
+}
+
+#pragma mark NSURLConnectionDelegate
+
+- (void)connection:(NSURLConnection *)connection didReceiveResponse:(NSURLResponse *)response {
+  NSHTTPURLResponse *httpResponse = (NSHTTPURLResponse *)response;
+  NSLog(@"status: %ld", httpResponse.statusCode);
+}
+
+- (void)connection:(NSURLConnection *)connection didReceiveData:(NSData *)data {
+}
+
+- (NSCachedURLResponse *)connection:(NSURLConnection *)connection
+                  willCacheResponse:(NSCachedURLResponse *)cachedResponse {
+  return nil;
+}
+
+- (void)connectionDidFinishLoading:(NSURLConnection *)connection {
+  NSLog(@"response complete");
+}
+
+- (void)connection:(NSURLConnection *)connection didFailWithError:(NSError *)error {
+  NSLog(@"ut oh");
+}
+
+@end

library/objective-c/BUILD

@@ -1,10 +1,18 @@
 licenses(["notice"])  # Apache 2
 
+load("@build_bazel_rules_apple//apple:ios.bzl", "ios_application")
+
 exports_files([
     "EnvoyEngine.h",
     "EnvoyTypes.h",
 ])
 
+filegroup(
+    name = "envoy_engine_hdrs",
+    srcs = ["EnvoyEngine.h"],
+    visibility = ["//visibility:public"],
+)
+
 objc_library(
     name = "envoy_engine_objc_lib",
     srcs = [
@@ -17,3 +25,28 @@ objc_library(
     visibility = ["//visibility:public"],
     deps = ["//library/common:envoy_main_interface_lib"],
 )
+
+objc_library(
+    name = "appmain",
+    srcs = [
+        "AppDelegate.h",
+        "AppDelegate.mm",
+        "main.mm",
+    ],
+    copts = ["-std=c++14"],
+    data = ["config.yaml"],
+    sdk_dylibs = [
+        "resolv.9",
+        "c++",
+    ],
+    deps = ["envoy_engine_objc_lib"],
+)
+
+ios_application(
+    name = "app",
+    bundle_id = "com.foo.bar",
+    families = ["iphone"],
+    infoplists = ["Info.plist"],
+    minimum_os_version = "11.0",
+    deps = ["appmain"],
+)

library/objective-c/EnvoyEngine.h

@@ -86,8 +86,8 @@
  */
 + (EnvoyStatus)resetStream:(EnvoyStream *)stream;
 
-/// Performs necessary setup after Envoy has initialized and started running.
-/// TODO: create a post-initialization callback from Envoy to handle this automatically.
++ (void)makeRequest;
+
 + (void)setupEnvoy;
 
 @end