Merge pull request scylladb#2 from HippoBaro/hb_web_socket

Hb web socket

README.md

@@ -32,8 +32,8 @@ Recommended hardware configuration for SeaStar
 
 * CPUs - As much as you need. SeaStar is highly friendly for multi-core and NUMA
 * NICs - As fast as possible, we recommend 10G or 40G cards. It's possible to use
-       1G to but you may be limited by their capacity.
-       In addition, the more hardware queue per cpu the better for SeaStar. 
+       1G too but you may be limited by their capacity.
+       In addition, the more hardware queue per cpu the better for SeaStar.
        Otherwise we have to emulate that in software.
 * Disks - Fast SSDs with high number of IOPS.
 * Client machines - Usually a single client machine can't load our servers.

apps/fair_queue_tester/fair_queue_tester.cc

@@ -33,6 +33,7 @@
 #include <iomanip>
 #include <random>
 
+using namespace seastar;
 using namespace std::chrono_literals;
 
 static auto random_seed = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch()).count();

apps/httpd/main.cc

@@ -19,16 +19,21 @@
  * Copyright 2015 Cloudius Systems
  */
 
+#include <cryptopp/sha.h>
+#include <cryptopp/base64.h>
 #include "http/httpd.hh"
 #include "http/handlers.hh"
+#include "http/websocket_handler.hh"
 #include "http/function_handlers.hh"
 #include "http/file_handler.hh"
 #include "apps/httpd/demo.json.hh"
 #include "http/api_docs.hh"
 
 namespace bpo = boost::program_options;
 
+using namespace seastar;
 using namespace httpd;
+using namespace websocket;
 
 class handl : public httpd::handler_base {
 public:
@@ -47,10 +52,59 @@ void set_routes(routes& r) {
     function_handler* h2 = new function_handler([](std::unique_ptr<request> req) {
         return make_ready_future<json::json_return_type>("json-future");
     });
+
+    auto ws_echo_handler = new websocket::ws_handler<SERVER>();
+
+    ws_echo_handler->on_message_future([] (const std::unique_ptr<request>& req, duplex_stream<SERVER>& stream,
+            message<SERVER> message) {
+        return stream.write(std::move(message)).then([&stream] {
+            return stream.flush();
+        });
+    });
+
+    auto ws_sha1_handler = new websocket::ws_handler<SERVER>();
+
+    ws_sha1_handler->on_connection_future([] (const std::unique_ptr<request>& req, duplex_stream<SERVER>& stream) {
+        return stream.write(websocket::message<SERVER>(TEXT, "Hello from seastar ! Send any message to this endpoint"
+                " and get back it's payload SHA1 in base64 format !")).then([&stream] {
+            return stream.flush();
+        });
+    });
+
+    ws_sha1_handler->on_message_future([] (const std::unique_ptr<request>& req, duplex_stream<SERVER>& stream,
+            message<SERVER> message) {
+        CryptoPP::SHA hash;
+        byte digest[CryptoPP::SHA::DIGESTSIZE];
+        hash.Update((byte*) message.payload.begin(), message.payload.size());
+        hash.Final(digest);
+
+        sstring base64;
+
+        CryptoPP::Base64Encoder encoder;
+        encoder.Put(digest, sizeof(digest));
+        encoder.MessageEnd();
+        CryptoPP::word64 size = encoder.MaxRetrievable();
+        if (size) {
+            base64.resize(size);
+            encoder.Get((byte*) base64.data(), base64.size());
+
+            return stream.write(websocket::message<SERVER>(TEXT, base64.substr(0, base64.size() - 1))).then([&stream] {
+                return stream.flush();
+            });
+        }
+        return make_ready_future();
+    });
+
+    ws_sha1_handler->on_disconnection([] (const std::unique_ptr<request>& req) {
+        print("websocket client disconnected\n");
+    });
+
     r.add(operation_type::GET, url("/"), h1);
     r.add(operation_type::GET, url("/jf"), h2);
-    r.add(operation_type::GET, url("/file").remainder("path"),
-            new directory_handler("/"));
+    r.add(operation_type::GET, url("/file").remainder("path"), new directory_handler("/"));
+    r.put("/", ws_echo_handler);
+    r.put("/sha1", ws_sha1_handler);
+
     demo_json::hello_world.set(r, [] (const_req req) {
         demo_json::my_object obj;
         obj.var1 = req.param.at("var1");
@@ -65,18 +119,18 @@ void set_routes(routes& r) {
 int main(int ac, char** av) {
     app_template app;
     app.add_options()("port", bpo::value<uint16_t>()->default_value(10000),
-            "HTTP Server port");
+                      "HTTP Server port");
     return app.run_deprecated(ac, av, [&] {
-        auto&& config = app.configuration();
+        auto &&config = app.configuration();
         uint16_t port = config["port"].as<uint16_t>();
         auto server = new http_server_control();
         auto rb = make_shared<api_registry_builder>("apps/httpd/");
         server->start().then([server] {
             return server->set_routes(set_routes);
-        }).then([server, rb]{
-            return server->set_routes([rb](routes& r){rb->set_api_doc(r);});
-        }).then([server, rb]{
-            return server->set_routes([rb](routes& r) {rb->register_function(r, "demo", "hello world application");});
+        }).then([server, rb] {
+            return server->set_routes([rb](routes &r) { rb->set_api_doc(r); });
+        }).then([server, rb] {
+            return server->set_routes([rb](routes &r) { rb->register_function(r, "demo", "hello world application"); });
         }).then([server, port] {
             return server->listen(port);
         }).then([server, port] {
@@ -85,6 +139,5 @@ int main(int ac, char** av) {
                 return server->stop();
             });
         });
-
     });
 }

apps/iotune/iotune.cc

@@ -44,9 +44,13 @@
 #include "core/aligned_buffer.hh"
 #include "util/defer.hh"
 
+using namespace seastar;
 using namespace std::chrono_literals;
 
+namespace seastar {
 bool filesystem_has_good_aio_support(sstring directory, bool verbose);
+}
+
 class iotune_manager;
 
 class iotune_timeout_exception : public std::exception {

apps/memcached/ascii.rl

@@ -25,6 +25,8 @@
 #include <algorithm>
 #include <functional>
 
+using namespace seastar;
+
 %%{
 
 machine memcache_ascii_protocol;

apps/memcached/memcache.cc

@@ -48,12 +48,13 @@
 #define VERSION "v1.0"
 #define VERSION_STRING PLATFORM " " VERSION
 
+using namespace seastar;
 using namespace net;
 
-namespace bi = boost::intrusive;
-
 namespace memcache {
 
+namespace bi = boost::intrusive;
+
 static constexpr double default_slab_growth_factor = 1.25;
 static constexpr uint64_t default_slab_page_size = 1UL*MB;
 static constexpr uint64_t default_per_cpu_slab_size = 0UL; // zero means reclaimer is enabled.

apps/memcached/memcached.hh

@@ -22,6 +22,8 @@
 
 namespace memcache {
 
+using namespace seastar;
+
 class item;
 class cache;