Remove all that nasty html

.gitignore

@@ -1,3 +1,4 @@
 build/
 .direnv/
-.vscode
+.vscode
+html

CMakeLists.txt

@@ -46,6 +46,12 @@ set_target_properties(fastinahurry PROPERTIES
   CXX_STANDARD_REQUIRED YES
 )
 
+add_executable(server ${SRC_FILES} server_main.cc)
+set_target_properties(server PROPERTIES
+  CXX_STANDARD 23
+  CXX_STANDARD_REQUIRED YES
+)
+
 # add_executable(fastinahurry ${SRC_FILES} main.cc)
 # add_executable(fastinahurry ${SRC_FILES} main.cc)
 
@@ -62,6 +68,21 @@ target_include_directories(fastinahurry PRIVATE
   ${CMAKE_CURRENT_BINARY_DIR} # For generated *.pb.h
 )
 
+target_link_libraries(server PRIVATE
+  trading_grpc_proto
+  nlohmann_json::nlohmann_json
+  ${_REFLECTION}
+  ${_GRPC_GRPCPP}
+  ${_PROTOBUF_LIBPROTOBUF}
+)
+
+target_include_directories(server PRIVATE 
+  ${CMAKE_SOURCE_DIR}/include
+  ${CMAKE_CURRENT_BINARY_DIR} # For generated *.pb.h
+)
+
+target_compile_features(server PRIVATE cxx_std_23)
+
 add_library(trading_grpc_proto
   ${trading_grpc_srcs}
   ${trading_grpc_hdrs}

include/Algo.hh

@@ -22,6 +22,7 @@ public:
     void generate_orders();
     void process();
     void start_background_processing();
+    void start_market_data_streaming();
     grpc::Status send(trading::Order&);
     bool initialized();
     void stop();

include/Publisher.hh

@@ -3,12 +3,22 @@
 #include "trading.grpc.pb.h"
 #include <grpcpp/grpcpp.h>
 #include <memory>
+#include <atomic>
+#include <thread>
 
 class Publisher {
 public:
     explicit Publisher(std::shared_ptr<grpc::Channel> channel);
 
     grpc::Status send_order(const trading::Order order);
+    void start_market_data_stream(const std::string& symbol);
+    void stop_market_data_stream();
+    
 private:
     std::unique_ptr<trading::TradingService::Stub> _stub;
+    std::atomic<bool> _streaming{false};
+    std::thread _stream_thread;
+    
+    void stream_market_data(const std::string& symbol);
+    trading::Order generate_order_from_market_data(const trading::MarketData& data);
 };

main.cc

@@ -11,7 +11,6 @@ int main(int argc, char* argv[]) {
     for (uint8_t i = 0; i < argc; i++) {
         std::printf("argument[%d]: %s\n", i, argv[i]);
     }
-    assert(argc == 2);
     Controller ctlr(argc, argv);
     ctlr.start();
 

server_main.cc

@@ -0,0 +1,20 @@
+#include <iostream>
+#include <sstream>
+
+void RunServer(std::string);
+void print_help() {
+    std::ostringstream ss;
+    ss << "Usage: ./build/server <port>" << std::endl;
+    std::cout << ss.str();
+}
+
+int main(int argc, char** argv) {
+    std::cout << "Starting gRPC market data server..." << std::endl;
+    if (argc != 2) {
+        print_help();
+        return 1;
+    };
+    RunServer(std::string(argv[1]));
+
+    return 0;
+}

src/Algo.cc

@@ -29,6 +29,10 @@ void Algo::stop() {
     _stopping.store(true, std::memory_order_release);
     _running.store(false, std::memory_order_release);
 
+    if (_pub) {
+        _pub->stop_market_data_stream();
+    }
+
     _reader_thread.request_stop();
     for (auto& t : _worker_threads) {
         t.request_stop();
@@ -84,6 +88,15 @@ void Algo::process() {
     }
 }
 
+void Algo::start_market_data_streaming() {
+    if (!_pub) {
+        std::cerr << "[Error] Publisher not initialized. Call initialize() first." << std::endl;
+        return;
+    }
+    
+    _pub->start_market_data_stream("AAPL");
+}
+
 void Algo::start_background_processing() {
     // Thread to read from file source and enqueue
     _reader_thread = std::jthread([this](std::stop_token stoken) {
@@ -96,7 +109,7 @@ void Algo::start_background_processing() {
         }
     });
 
-    std::size_t num_workers = 2;
+    std::size_t num_workers = 1;
     for (std::size_t i = 0; i < num_workers; ++i) {
         _worker_threads.emplace_back([this] (std::stop_token stoken) {
             while(!stoken.stop_requested()) {

src/Controller.cc

@@ -11,10 +11,12 @@ Controller::Controller(int argc, char* argv[])
 
 void Controller::start() {
     if (!_algo->initialized()) _algo->initialize();
+    
+    _algo->start_market_data_streaming();
+    
     while(_algo->is_running()) {
-        _algo->process();
+        std::this_thread::sleep_for(std::chrono::seconds(1));
-        std::this_thread::sleep_for(std::chrono::seconds(3));
         _count++;
-        if (_count > 5) _algo->stop();
+        if (_count >= 10) _algo->stop();
     }
 }

src/Publisher.cc

@@ -1,7 +1,7 @@
 #include "Publisher.hh"
 #include <iostream>
 #include <thread>
-
+#include <random>
 
 Publisher::Publisher(std::shared_ptr<grpc::Channel> channel) 
     : _stub(trading::TradingService::NewStub(channel)) {}
@@ -19,17 +19,78 @@ grpc::Status Publisher::send_order(const trading::Order order) {
     return status;
 }
 
-// void StreamMarketData() {
+void Publisher::start_market_data_stream(const std::string& symbol) {
-//     MarketRequest req;
+    if (_streaming.load()) {
-//     req.set_symbol("AAPL");
+        std::cout << "Market data stream already running" << std::endl;
+        return;
+    }
+    
+    _streaming.store(true);
+    _stream_thread = std::thread(&Publisher::stream_market_data, this, symbol);
+    std::cout << "Started market data stream for " << symbol << std::endl;
+}
 
-//     ClientContext context;
+void Publisher::stop_market_data_stream() {
-//     std::unique_ptr<grpc::ClientReader<MarketData>> reader(
+    if (!_streaming.load()) {
-//         stub_->StreamMarketData(&context, req));
+        return;
+    }
+    
+    _streaming.store(false);
+    if (_stream_thread.joinable()) {
+        _stream_thread.join();
+    }
+    std::cout << "Stopped market data stream" << std::endl;
+}
 
-//     MarketData data;
+void Publisher::stream_market_data(const std::string& symbol) {
-//     while (reader->Read(&data)) {
+    trading::MarketRequest req;
-//     std::cout << "Market: " << data.symbol() << " $" << data.price()
+    req.set_symbol(symbol);
-//                 << " @ " << data.timestamp() << std::endl;
+
-//     }
+    grpc::ClientContext context;
-// }
+    std::unique_ptr<grpc::ClientReader<trading::MarketData>> reader(
+        _stub->StreamMarketData(&context, req));
+
+    trading::MarketData data;
+    while (reader->Read(&data) && _streaming.load()) {
+        std::cout << "Received market data: " << data.symbol() 
+                  << " @ $" << data.price() 
+                  << " (timestamp: " << data.timestamp() << ")" << std::endl;
+        
+        trading::Order order = generate_order_from_market_data(data);
+        grpc::Status status = send_order(order);
+        
+        if (!status.ok()) {
+            std::cerr << "[Error] Failed to send order: " << status.error_message() << std::endl;
+        }
+    }
+    
+    grpc::Status status = reader->Finish();
+    if (!status.ok()) {
+        std::cerr << "[Error] Market data stream ended with error: " << status.error_message() << std::endl;
+    }
+    
+    _streaming.store(false);
+}
+
+trading::Order Publisher::generate_order_from_market_data(const trading::MarketData& data) {
+    trading::Order order;
+    order.set_symbol(data.symbol());
+    
+    static std::random_device rd;
+    static std::mt19937 gen(rd());
+    static std::uniform_int_distribution<> quantity_dist(10, 100);
+    static std::uniform_int_distribution<> side_dist(1, 2);
+    static std::uniform_real_distribution<> price_adjustment(-2.0, 2.0);
+    
+    order.set_quantity(quantity_dist(gen));
+    order.set_side(static_cast<trading::Side>(side_dist(gen)));
+    
+    double adjusted_price = data.price() + price_adjustment(gen);
+    order.set_price(adjusted_price);
+    
+    std::cout << "Generated order: " << (order.side() == trading::BUY ? "BUY" : "SELL")
+              << " " << order.quantity() << " " << order.symbol() 
+              << " @ $" << order.price() << std::endl;
+    
+    return order;
+}

src/grpc_server.cc

@@ -2,6 +2,9 @@
 #include <grpcpp/grpcpp.h>
 #include <iostream>
 #include <thread>
+#include <cmath>
+#include <cstdlib>
+#include <ctime>
 
 using grpc::Server;
 using grpc::ServerBuilder;
@@ -26,30 +29,48 @@ class TradingServiceImpl final : public TradingService::Service {
 
     Status StreamMarketData(ServerContext* ctx, const MarketRequest* req,
                             grpc::ServerWriter<MarketData>* writer) override {
+        std::cout << "Starting market data stream for symbol: " << req->symbol() << std::endl;
         
-            uint32_t max_market_streams = 5;
+        uint32_t counter = 0;
-            for (uint32_t i = 0; i < max_market_streams; i++) {
+        while (!ctx->IsCancelled()) {
-                MarketData data;
+            MarketData data;
-                data.set_symbol(req->symbol());
+            data.set_symbol(req->symbol());
-                data.set_price(100 + i);
+            
-                data.set_timestamp(time(nullptr));
+            double base_price = 100.0;
-                writer->Write(data);
+            double price_variation = 5.0 * sin(counter * 0.1) + (rand() % 10 - 5) * 0.1;
-                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            data.set_price(base_price + price_variation);
+            data.set_timestamp(time(nullptr));
+            
+            if (!writer->Write(data)) {
+                break;
             }
-            return Status::OK;
+            
+            std::cout << "Sent market data: " << req->symbol() 
+                     << " @ $" << data.price() << std::endl;
+            
+            counter++;
+            std::this_thread::sleep_for(std::chrono::seconds(3));
+        }
+        
+        std::cout << "Market data stream ended for: " << req->symbol() << std::endl;
+        return Status::OK;
     }
 };
 
-    void RunServer() {
+    void RunServer(std::string port) {
-        std::string addr("0.0.0.0:50051");
+        std::string addr("0.0.0.0");
+        std::string socket = addr;
+        std::ostringstream ss;
+        ss << addr << ":" << port;
         TradingServiceImpl service;
 
         ServerBuilder builder;
-        builder.AddListeningPort(addr, grpc::InsecureServerCredentials());
+
+        builder.AddListeningPort(ss.str(), grpc::InsecureServerCredentials());
         builder.RegisterService(&service);
         std::unique_ptr<Server> server(builder.BuildAndStart());
 
-        std::cout << "Market server listening on" << addr << std::endl;
+        std::cout << "Market server listening on " << addr << ":" << port << std::endl;
         
         server->Wait();
     }