#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();
  });
}


#include "bhome_msg.pb.h"
#include "bhome_msg_api.pb.h"
using namespace bhome_msg;

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

  // test_rr();
  // test_ps();
  // return 0;

  int reply = 1;

  string id("hello-reply");
  if (argc > 1) {
    printf("this is request\n");
    id = "hello-request";
    reply = 0;
  }else{
    printf("this is reply\n");
  }

  ProcInfo pi;
  pi.set_proc_id(id);
  pi.set_name("works");

  string out;
  pi.SerializeToString(&out);

  void* rep;
  int repl = 0;
  BHRegister(out.data(), out.size(), &rep, &repl, 500);

  if (reply){
    while (true) {
      void* pid;
      int pidl;
      void* req;
      int reql;
      void* src;

      if (BHReadRequest(&pid, &pidl, &req, &reql, &src, 500)){

        bhome_msg::MsgRequestTopic msg;
        msg.ParseFromArray(req, reql);
        printf("recv request %d msg data %s\n", reql, msg.data().c_str());

        bhome_msg::MsgRequestTopicReply rep;
        rep.set_data(msg.data() + "-reply");
        string srep;
        rep.SerializeToString(&srep);

        // auto s = chrono::steady_clock::now();
        int ret = BHSendReply(src, srep.data(), srep.size());
        // auto e = chrono::steady_clock::now();
        // printf("reply time %lu ms\n", chrono::duration_cast<chrono::milliseconds>(e-s).count());

      }else{
        // usleep(50000);
        // printf("BHReadRequest no data\n");
      }
    }
  }else{
    printf("start request %d\n", atoi(argv[1]));
    bhome_msg::BHAddress addr;
    addr.set_ip("192.168.20.108");
    addr.set_port(atoi(argv[1]));

    string node;
    addr.SerializeToString(&node);

    void* pid;
    int pidl;
    void* rep;
    int repl;
    unsigned idx = 0;
    while(1){

      bhome_msg::MsgRequestTopic msg;
      msg.set_topic("hello-reply");
      msg.set_data("hell-world-" + to_string(getpid()) + "-" + to_string(idx++));
      string smsg;
      msg.SerializeToString(&smsg);

      // auto s = chrono::steady_clock::now();
      int ret = BHRequest(node.data(), node.size(), smsg.data(), smsg.size(),
        &pid, &pidl, &rep, &repl, 5000);
      // auto e = chrono::steady_clock::now();
      // printf("request time %lu ms\n", chrono::duration_cast<chrono::milliseconds>(e-s).count());

      bhome_msg::MsgRequestTopicReply prep;
      prep.ParseFromArray(rep, repl);

      printf("pid %d BHRequest rep data %s size %d ret %d\n", getpid(), prep.data().c_str(),
        repl, ret);

    }
  }

  return 0;
}