fdbus实现demo (包括client端和server端)

/* * Copyright (C) 2015 Jeremy Chen jeremy_cz@yahoo.com * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #define FDB_LOG_TAG "FDB_TEST_CLIENT" #include <common_base/fdbus.h> #include <common_base/cJSON/cJSON.h> #include <common_base/CFdbCJsonMsgBuilder.h> #include <common.base.mcuipcfdbus.pb.h> #include <CFdbProtoMsgBuilder.h> #include <iostream> #define FDB_INVOKE_SYNC 1 static CBaseWorker main_worker; class CMediaClient; void printMetadata(FdbObjectId_t obj_id, const CFdbMsgMetadata &metadata) { uint64_t time_c2s; uint64_t time_s2r; uint64_t time_r2c; uint64_t time_total; CFdbMessage::parseTimestamp(metadata, time_c2s, time_s2r, time_r2c, time_total); FDB_LOG_I("OBJ %d , client->server: %llu, arrive->reply: %llu, reply->receive: %llu, total: %llu\n", obj_id, time_c2s, time_s2r, time_r2c, time_total); } /* a timer sending request to server periodically */ class CInvokeTimer : public CMethodLoopTimer<CMediaClient> { public: CInvokeTimer(CMediaClient *client); }; class CMediaClient : public CBaseClient { public: CMediaClient(const char *name, CBaseWorker *worker = 0) : CBaseClient(name, worker) { /* create the request timer, attach to a worker thread, but do not start it */ mTimer = new CInvokeTimer(this); mTimer->attach(&main_worker, false); } /* called periodically in timer to get song information */ void callServer(CMethodLoopTimer<CMediaClient> *timer) { ::mcugateway::FdbSendMessageRequest req_msg; std::string str("hello"); req_msg.set_priority(::mcugateway::FdbSendMessageRequest::LOW); req_msg.set_appdata(str); CBaseJob::Ptr ref(new CBaseMessage(::mcugateway::CodeId::SEND_MESSAGE)); CFdbProtoMsgBuilder builder(req_msg); invoke(ref, builder); auto res_msg = castToMessage<CBaseMessage *>(ref); if (res_msg->isStatus()) { int32_t id; std::string reason; if (!res_msg->decodeStatus(id, reason)) { FDB_LOG_E("onReply: fail to decode status!\n"); return; } if (res_msg->isError()) { FDB_LOG_I("sync reply: status is received: msg code: %d, id: %d, reason: %s\n", res_msg->code(), id, reason.c_str()); } return; } ::mcugateway::FdbSendMessageResponse recv_msg; CFdbProtoMsgParser parser(recv_msg); if(res_msg->deserialize(parser)){ FDB_LOG_I("client recv return number: %d\n", (int)recv_msg.returnenum()); } } protected: /* called when connected to the server */ void onOnline(FdbSessionId_t sid, bool is_first) { FDB_LOG_I("client session online: %d\n", sid); if(is_first){ CFdbMsgSubscribeList subscribe_list; addNotifyItem(subscribe_list, ::mcugateway::CodeId::BROADCAST_MCU_TO_CAN); addNotifyItem(subscribe_list, ::mcugateway::CodeId::BROADCAST_MCU_TO_POWER); addNotifyItem(subscribe_list, ::mcugateway::CodeId::BROADCAST_MCU_TO_EOL); subscribe(subscribe_list); } } /* called when disconnected from server */ void onOffline(FdbSessionId_t sid, bool is_last) { } /* called when events broadcasted from server is received */ void onBroadcast(CBaseJob::Ptr &msg_ref) { auto msg = castToMessage<CBaseMessage *>(msg_ref); FDB_LOG_I("Broadcast is received: %d; filter: %s\n", msg->code(), msg->getFilter()); switch (msg->code()) { case ::mcugateway::CodeId::BROADCAST_MCU_TO_CAN: { FDB_LOG_I("client recv power broad msg"); } break; case ::mcugateway::CodeId::BROADCAST_MCU_TO_POWER: { FDB_LOG_I("client recv eol broad msg"); } break; case ::mcugateway::CodeId::BROADCAST_MCU_TO_EOL: { FDB_LOG_I("client recv mcu broad msg"); } break; default: break; } } /* called when client call asynchronous version of invoke() and reply() is called at server */ void onReply(CBaseJob::Ptr &msg_ref) { auto msg = castToMessage<CBaseMessage *>(msg_ref); FDB_LOG_I("response is receieved. sn: %d\n", msg->sn()); /* print performance statistics */ CFdbMsgMetadata md; msg->metadata(md); printMetadata(objId(), md); switch (msg->code()) { case 0: { if (msg->isStatus()) { /* Unable to get intended reply from server... Check what happen. */ if (msg->isError()) { int32_t error_code; std::string reason; if (!msg->decodeStatus(error_code, reason)) { FDB_LOG_E("onReply: fail to decode status!\n"); return; } FDB_LOG_I("onReply(): status is received: msg code: %d, error_code: %d, reason: %s\n", msg->code(), error_code, reason.c_str()); } return; } /* * recover protocol buffer from incoming package * it should match the type replied from server */ } break; default: break; } } /* check if something happen... */ void onStatus(CBaseJob::Ptr &msg_ref , int32_t error_code , const char *description) { auto msg = castToMessage<CBaseMessage *>(msg_ref); if (msg->isSubscribe()) { if (msg->isError()) { } else { FDB_LOG_I("subscribe is ok! sn: %d is received.\n", msg->sn()); } } FDB_LOG_I("Reason: %s\n", description); } private: CInvokeTimer *mTimer; }; /* create a timer: interval is 1000ms; cyclically; when timeout, call CMediaClient::callServer() */ CInvokeTimer::CInvokeTimer(CMediaClient *client) : CMethodLoopTimer<CMediaClient>(1000, true, client, &CMediaClient::callServer) {} int main(int argc, char **argv) { /* start fdbus context thread */ FDB_CONTEXT->start(); CBaseWorker *worker_ptr = &main_worker; /* start worker thread */ worker_ptr->start(); /* create clients and connect to address: svc://service_name */ std::string server_name = "fdbus_mcu_ipc"; std::string url(FDB_URL_SVC); url += server_name; server_name += "_client"; std::cout << "connect server name: " << url << std::endl; auto client = new CMediaClient(server_name.c_str(), worker_ptr); client->enableReconnect(true); client->connect(url.c_str()); /* convert main thread into worker */ CBaseWorker background_worker; background_worker.start(FDB_WORKER_EXE_IN_PLACE); }