103 lines
3.5 KiB
C++
103 lines
3.5 KiB
C++
#include "ApiServerInterop.h"
|
|
#include "lib/logging/logging.h"
|
|
#include "lib/quickbus/quickbus.h"
|
|
#include "lib/utils/bts_assert.h"
|
|
|
|
namespace bts {
|
|
namespace api {
|
|
|
|
// defined in ApiClientInterop.h
|
|
std::string base_topic(const char *type_name, const char *service_qualifier);
|
|
bts::qbus::RPCServiceName rpc_topic(const std::string &base_topic);
|
|
bts::qbus::Topic stream_topic(
|
|
const std::string &base_topic, const char *method, const char *qualifier);
|
|
|
|
struct ApiServerCore {
|
|
ApiServerCore(
|
|
Quickbus &qbus,
|
|
const std::string topic,
|
|
const qbus::OnRPCRequest &handler)
|
|
: qbus(qbus), topic(topic), rpc(qbus, rpc_topic(topic), handler) {}
|
|
|
|
Quickbus &qbus;
|
|
std::string topic;
|
|
qbus::RPCServer rpc;
|
|
};
|
|
|
|
CAPI ApiServerCore *api_server_create(
|
|
Quickbus *pbus,
|
|
const char *type_name,
|
|
const char *service_qualifier,
|
|
raw_rpc_request_callback cb,
|
|
void *extra) {
|
|
bts_assert(pbus);
|
|
// Invalid characters in qualifier: +
|
|
for (auto ch : {'+'}) {
|
|
bts_assert(strchr(service_qualifier, ch) == nullptr);
|
|
}
|
|
auto service_name = base_topic(type_name, service_qualifier);
|
|
LOG(Api, Info, "Create server: {}", service_name);
|
|
return new ApiServerCore(
|
|
*pbus, service_name,
|
|
[=](qbus::RPCRequestPtr const &req, qbus::ReplyHandler const &reply) {
|
|
// Extend the lifetime of the ReplyHandler by moving it to the heap
|
|
qbus::ReplyHandler *phandler =
|
|
new qbus::ReplyHandler(std::move(reply));
|
|
LOG(Api, Diag, "Received rpc request: {}",
|
|
reinterpret_cast<intptr_t>(phandler));
|
|
cb((uint8_t *)(req->value().c_str()), req->value().size(),
|
|
reinterpret_cast<ReplyHandler *>(phandler), extra);
|
|
});
|
|
}
|
|
|
|
CAPI void api_server_call_callback(
|
|
ReplyHandler **ppvhandler, const uint8_t *buffer, int length) {
|
|
bts_assert(ppvhandler);
|
|
bts_assert(*ppvhandler);
|
|
// LOG(Api,Diag,"Send rpc reply:
|
|
// {}",reinterpret_cast<intptr_t>(*ppvhandler));
|
|
qbus::ReplyHandler **pphandler =
|
|
reinterpret_cast<qbus::ReplyHandler **>(ppvhandler);
|
|
std::string reply_buffer((const char *)buffer, (size_t)length);
|
|
qbus::RPCReplyPtr replyptr =
|
|
std::make_shared<qbus::RPCReply>(std::move(reply_buffer));
|
|
(**pphandler)(replyptr);
|
|
delete (*pphandler);
|
|
*pphandler = nullptr;
|
|
}
|
|
|
|
CAPI TableHandle api_server_create_table(
|
|
ApiServerCore *pserver, const char *method, const char *qualifier) {
|
|
bts_assert(pserver);
|
|
auto topic = stream_topic(pserver->topic, method, qualifier);
|
|
LOG(Api, Info, "Create table: {}", topic.value());
|
|
return new qbus::TablePublisher(pserver->qbus, topic);
|
|
}
|
|
|
|
CAPI void api_server_update_table(
|
|
TableHandle htable, StreamUpdate *raw_changes, int change_count) {
|
|
bts_assert(htable);
|
|
qbus::TablePublisher *ptable =
|
|
reinterpret_cast<qbus::TablePublisher *>(htable);
|
|
std::vector<qbus::Change> changes(change_count);
|
|
for (int x = 0; x < change_count; ++x) {
|
|
StreamUpdate &ch = raw_changes[x];
|
|
std::string buffer(ch.buffer, ch.buffer_len);
|
|
qbus::RecordPtr record =
|
|
std::make_shared<qbus::Record>(std::move(buffer));
|
|
changes[x] = qbus::Change(
|
|
(qbus::Change::Type)ch.type, qbus::RecordKey(ch.key), record);
|
|
}
|
|
ptable->publish(std::move(changes));
|
|
}
|
|
|
|
CAPI void api_server_destroy_table(qbus::TablePublisher *ptable) {
|
|
if (ptable) {
|
|
ptable->disconnect();
|
|
delete ptable;
|
|
}
|
|
}
|
|
|
|
} // namespace api
|
|
} // namespace bts
|