#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <vector>
#include <thread>
#include <chrono>
#include <atomic>
using namespace std;

#include "src/bn_api.h"

static void test_rr(){

  thread([]{
    string base_cont("test_req_rep==");

    atomic<uint64_t> index{0};
    vector<thread> v_t;
    for (int i = 0; i < 621; i++){
      v_t.emplace_back([&base_cont, i, &index]{

        while (true) {
          // printf("start request\n");
          // auto s = chrono::steady_clock::now();
          auto msg("[Thread("+to_string(i)+")]->"+base_cont+to_string(index++));
          TestRequest(0, msg.c_str(), msg.length());
          this_thread::sleep_for(chrono::milliseconds(10));
          // auto e = chrono::steady_clock::now();
          // printf("======>>thread %d TestRequest time %ld ms\n", i, chrono::duration_cast<chrono::milliseconds>(e-s).count());
        }
      });
    }

    while (true) {
      // printf("start request\n");
      // auto s = chrono::steady_clock::now();
      auto msg(base_cont+to_string(index++));
      TestRequest(0, msg.c_str(), msg.length());
      this_thread::sleep_for(chrono::milliseconds(10));
      // auto e = chrono::steady_clock::now();
      // printf("TestRequest time %ld ms\n", chrono::duration_cast<chrono::milliseconds>(e-s).count());
    }
  }).detach();

    while(true){
      TestReply(0, -1);
    }

}

static void test_ps(){

  const string t("topics_");
  vector<string> topics;
  for(int i = 0; i < 3; i++){
    topics.emplace_back(t + to_string(i+1));
  }

  string base_cont("test_pub_sub==");
  // while (base_cont.size() < 12662) {
  //   base_cont += base_cont;
  // }
  thread([&]{
    this_thread::sleep_for(chrono::seconds(3));
    while (true) {
      for(auto && i : topics){
        auto msg = base_cont + "test_ps pub message "+i+"-->msg";
        TestPub(i.c_str(), i.length(), msg.c_str(), msg.length());
        this_thread::sleep_for(chrono::milliseconds{126});
      }
    }
  }).detach();

  for(auto && i : topics){
    TestSub(i.c_str(), i.length(), 0, 0);
  }
  // this_thread::sleep_for(chrono::seconds(3));
  while (true) {
    char *msg;
    int msg_len;
    TestSub(NULL, 0, (void**)&msg, &msg_len);
    this_thread::sleep_for(chrono::seconds{1});
  }
}

vector<thread> v_t;
template<class F>
void run_test(F&& f){
  v_t.emplace_back([f]{
    f();
  });
}

int main(int argc, char const *argv[])
{
  // run_test([&]{test_rr();});
  test_rr();
  test_ps();

  return 0;
}