feat add rx test

CMakeLists.txt

@@ -32,7 +32,7 @@ if (SLED_LOCATION_PATH)
     target_compile_definitions(sled PRIVATE __SLED_LOCATION_PATH="${SLED_LOCATION_PATH}")
 endif()
 # add_subdirectory(3party/eigen EXCLUDE_FROM_ALL)
-target_include_directories(sled PUBLIC 3party/eigen 3party/inja 3party/rxcpp)
+target_include_directories(sled PUBLIC include 3party/eigen 3party/inja 3party/rxcpp)
 target_sources(
   sled
   PRIVATE 

include/sled/rx.h

@@ -1,11 +1,45 @@
 #pragma once
+
 #ifndef SLED_RX_H
 #define SLED_RX_H
-
+// #include "rx-scheduler.hpp"
-#include "rx-lite.hpp"
+// #include "rx-lite.hpp"
+#include "rx-includes.hpp"
+#include "sled/system/fiber/scheduler.h"
+#include "sled/time_utils.h"
 
 namespace sled {
 using namespace rxcpp;
-}
+
+// struct fiber_scheduler : public schedulers::scheduler_interface {
+//     class new_worker : public schedulers::worker_interface {
+//     public:
+//         new_worker(composite_subscription cs) {}
+//
+//         ~new_worker() override {}
+//
+//         clock_type::time_point now() const override { return clock_type::now(); };
+//
+//         void schedule(const sled::schedulers::schedulable &scbl) const override { return schedule(now(), scbl); };
+//
+//         void schedule(clock_type::time_point when, const sled::schedulers::schedulable &scbl) const override{};
+//     };
+//
+// public:
+//     ~fiber_scheduler() override {}
+//
+//     clock_type::time_point now() const override
+//     {
+//         // auto ns = sled::TimeNanos();
+//         // return clock_type::time_point(std::chrono::nanoseconds(ns));
+//         return clock_type::now();
+//     }
+//
+//     sled::schedulers::worker create_worker(composite_subscription cs) const override
+//     {
+//         return sled::schedulers::worker(cs, std::make_shared<new_worker>());
+//     };
+// };
+}// namespace sled
 
 #endif//  SLED_RX_H

src/rx_test.cc

@@ -1,10 +1,17 @@
-#include "schedulers/rx-currentthread.hpp"
+#include <chrono>
 #include <gtest/gtest.h>
 #include <sled/rx.h>
+#include <sled/time_utils.h>
 
 TEST(RX, interval)
 {
-    auto sc = sled::schedulers::make_current_thread();
+    // 2 + 4 + 6 = 12
-    sled::observable<>::interval(std::chrono::seconds(1), sled::synchronize_in_one_worker(sc))
+    sled::observable<>::interval(std::chrono::milliseconds(300))
-        .subscribe([](long counter) {});
+        .subscribe_on(rxcpp::synchronize_new_thread())
+        .observe_on(rxcpp::observe_on_new_thread())
+        .map([](long value) { return value * 2; })
+        .take(3)
+        .reduce(0, [](int acc, int value) { return acc + value; })
+        .as_blocking()
+        .subscribe([=](long counter) { EXPECT_EQ(counter, 12); }, []() {});
 }

src/strings/base64.cc

@@ -1,25 +1,19 @@
+#include "sled/strings/base64.h"
 #include <fmt/format.h>
-#include <sled/strings/base64.h>
 #include <sstream>
 
 namespace sled {
-const char kBase64Chars[] =
+const char kBase64Chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
-    "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
 const int kInvBase64Chars[] = {
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, 62, -1, -1, -1, 63, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61,
-    -1, -1, -1, -1, -1, 62, -1, -1, -1, 63, 52, 53, 54, 55, 56, 57, 58, 59, 60,
+    -1, -1, -1, -1, -1, -1, -1, 0,  1,  2,  3,  4,  5,  6,  7,  8,  9,  10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21,
-    61, -1, -1, -1, -1, -1, -1, -1, 0,  1,  2,  3,  4,  5,  6,  7,  8,  9,  10,
+    22, 23, 24, 25, -1, -1, -1, -1, -1, -1, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44,
-    11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, -1, -1, -1, -1,
+    45, 46, 47, 48, 49, 50, 51, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    43, 44, 45, 46, 47, 48, 49, 50, 51, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
+    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1,
-    -1, -1, -1, -1, -1, -1, -1, -1, -1,
 };
 
 inline bool
@@ -32,8 +26,7 @@ std::string
 Base64::Encode(const std::string &input)
 {
     std::stringstream ss;
-    const unsigned char *data =
+    const unsigned char *data = reinterpret_cast<const unsigned char *>(input.c_str());
-        reinterpret_cast<const unsigned char *>(input.c_str());
     int value = 0;
     int value_bits = 0;
     for (unsigned char c : input) {
@@ -50,9 +43,7 @@ Base64::Encode(const std::string &input)
      * 2           ->  4  -> (8 - value_bits - 2)
      * 4           ->  2  -> (8 - value_bits - 2)
      **/
-    if (value_bits > 0) {
+    if (value_bits > 0) { ss << kBase64Chars[((value << 8) >> (value_bits + 2)) & 0x3F]; }
-        ss << kBase64Chars[((value << 8) >> (value_bits + 2)) & 0x3F];
-    }
     while (ss.str().size() % 4) { ss << '='; }
 
     return ss.str();
@@ -79,10 +70,8 @@ Base64::Decode(const std::string &input)
             break;
         } else {
             // invalid character
-            return make_status_or<std::string>(
+            return make_status_or<std::string>(StatusCode::kInvalidArgument,
-                StatusCode::kInvalidArgument,
+                                               fmt::format("Invalid character [{}] at index [{}]", (char) c, index));
-                fmt::format("Invalid character [{}] at index [{}]", (char) c,
-                            index));
         }
         ++index;
     }

src/system/thread_pool_test.cc

@@ -4,13 +4,13 @@
 
 std::random_device rd;
 std::mt19937 mt(rd());
-std::uniform_int_distribution<int> dist(-10, 10);
+std::uniform_int_distribution<int> dist(-5, 5);
 auto rnd = std::bind(dist, mt);
 
 void
 simulate_hard_computation()
 {
-    std::this_thread::sleep_for(std::chrono::milliseconds(20 + rnd()));
+    // std::this_thread::sleep_for(std::chrono::milliseconds(10 + rnd()));
 }
 
 // Simple function that adds multiplies two numbers and prints the result