add different log info

include/aca_message_pulsar_consumer.h

@@ -26,6 +26,7 @@
 
 using namespace pulsar;
 using std::string;
+using std::vector;
 
 namespace aca_message_pulsar
 {
@@ -37,7 +38,7 @@ class ACA_Message_Pulsar_Consumer {
     string unicast_subscription_name; // Subscription name of the unicast pulsar consumer
 
     string multicast_topic_name; //A string representation of the topic to be consumed, for example: /hostid/00000000-0000-0000-0000-000000000000/netwconf/
-    string unicast_topic_name; 
+    vector<string> unicast_topic_name = vector<string>(); 
 
     ConsumerConfiguration multicast_consumer_config; //Configuration of the mulitcast pulsar consumer
     ConsumerConfiguration unicast_consumer_config; //Configuration of the unicast pulsar consumer
@@ -48,8 +49,6 @@ class ACA_Message_Pulsar_Consumer {
     Consumer multicast_consumer;
     Consumer unicast_consumer;
 
-    static string empty_topic;
-
   private:
     void setMulticastSubscriptionName(string subscription_name);
 
@@ -88,11 +87,9 @@ class ACA_Message_Pulsar_Consumer {
 
     bool unicastConsumerDispatched(int stickyHash);
 
-    bool unicastUnsubcribe();
-
-    bool unicastResubscribe(string topic, int stickyHash);
+    bool unicastUnsubscribeAll();
 
-    bool unicastResubscribe(bool isUnSubscribe, string topic="", string stickHash="");
+    bool unicastResubscribe(bool unSubscribe, string topic="", string stickHash="");
 };
 
 } // namespace aca_message_pulsar

src/comm/aca_grpc.cpp

@@ -174,7 +174,8 @@ void GoalStateProvisionerAsyncServer::ProcessPushGoalStatesStreamAsyncCall(
           //  process goalstateV2 in the thread pool
           //  It has read from the stream, now to GoalStateV2 should not be empty
           //  and we need to process it.
-          ACA_LOG_DEBUG("%s\n", "This call has already read from the stream, now we process the gsv2...");
+          ACA_LOG_DEBUG("%s\n", "ACA_GRPC: This call has already read from the stream, now we process the gsv2...");
+
 
           if (streamingCall->goalStateV2_.neighbor_states_size() == 1) {
             // if there's only one neighbor state, it means that it is pushed
@@ -189,16 +190,17 @@ void GoalStateProvisionerAsyncServer::ProcessPushGoalStatesStreamAsyncCall(
           }
           std::chrono::_V2::steady_clock::time_point start =
                   std::chrono::steady_clock::now();
+          ACA_LOG_INFO("%s\n", "ACA_GRPC: Start updating GoalStateV2...");
           int rc = Aca_Comm_Manager::get_instance().update_goal_state(
                   streamingCall->goalStateV2_, streamingCall->gsOperationReply_);
           if (rc == EXIT_SUCCESS) {
-            ACA_LOG_INFO("Control Fast Path streaming - Successfully updated host with latest goal state %d.\n",
+            ACA_LOG_INFO("ACA_GRPC: Control Fast Path streaming - Successfully updated host with latest goal state %d.\n",
                          rc);
           } else if (rc == EINPROGRESS) {
-            ACA_LOG_INFO("Control Fast Path streaming - Update host with latest goal state returned pending, rc=%d.\n",
+            ACA_LOG_INFO("ACA_GRPC: Control Fast Path streaming - Update host with latest goal state returned pending, rc=%d.\n",
                          rc);
           } else {
-            ACA_LOG_ERROR("Control Fast Path streaming - Failed to update host with latest goal state, rc=%d.\n",
+            ACA_LOG_ERROR("ACA_GRPC: Control Fast Path streaming - Failed to update host with latest goal state, rc=%d.\n",
                           rc);
           }
           std::chrono::_V2::steady_clock::time_point end =

src/comm/aca_message_pulsar_consumer.cpp

@@ -30,32 +30,32 @@ using pulsar::StickyRange;
 
 namespace aca_message_pulsar
 {
-string aca_message_pulsar::ACA_Message_Pulsar_Consumer::empty_topic="";
-
 void listener(Consumer consumer, const Message& message){
   alcor::schema::GoalStateV2 deserialized_GoalState;
   alcor::schema::GoalStateOperationReply gsOperationalReply;
   int rc;
   Result result;
 
-  ACA_LOG_DEBUG("\n<=====incoming message: %s\n",
-                message.getDataAsString().c_str());
-
+  ACA_LOG_INFO("%s","ACA_PULSAR_MQ: Successfully received the incoming message.\n");
+  ACA_LOG_INFO("%s","ACA_PULSAR_MQ: Start deserializing the message to GoalState...\n");
+  ACA_LOG_INFO("%s","<====================================================>\n");
   rc = Aca_Comm_Manager::get_instance().deserialize(
           (unsigned char *)message.getData(), message.getLength(), deserialized_GoalState);
+  ACA_LOG_INFO("%s","<====================================================>\n");
   if (rc == EXIT_SUCCESS) {
+    ACA_LOG_INFO("%s","ACA_PULSAR_MQ: Start updating GoalState...\n");
+    ACA_LOG_INFO("%s","<====================================================>\n");
     rc = Aca_Comm_Manager::get_instance().update_goal_state(
                 deserialized_GoalState, gsOperationalReply);
-
-
+    ACA_LOG_INFO("%s","<====================================================>\n");
     if (rc != EXIT_SUCCESS) {
-      ACA_LOG_ERROR("Failed to update host with latest goal state, rc=%d.\n", rc);
+      ACA_LOG_ERROR("ACA_PULSAR_MQ: ERROR, failed to update host with latest goal state, rc=%d.\n", rc);
     } else {
-      ACA_LOG_INFO("Successfully updated host with latest goal state %d.\n", rc);
+      ACA_LOG_INFO("ACA_PULSAR_MQ: Successfully updated host with latest goal state, rc=%d.\n", rc);
     }
 
   } else {
-    ACA_LOG_ERROR("Deserialization failed with error code %d.\n", rc);
+    ACA_LOG_ERROR("ACA_PULSAR_MQ: ERROR, failed to deserialize the message with error code %d.\n", rc);
   }
 
   // Now acknowledge message
@@ -78,11 +78,11 @@ void ACA_Message_Pulsar_Consumer::init(string topic, string brokers, string subs
     setUnicastSubscriptionName(subscription_name);
     setMulticastSubscriptionName(subscription_name);
 
-    ACA_LOG_DEBUG("Broker list: %s\n", this->brokers_list.c_str());
-    ACA_LOG_DEBUG("Unicast consumer topic name: %s\n", this->unicast_topic_name.c_str());
-    ACA_LOG_DEBUG("Unicast consumer subscription name: %s\n", this->unicast_subscription_name.c_str());
-    ACA_LOG_DEBUG("Multicast consumer topic name: %s\n", this->multicast_topic_name.c_str());
-    ACA_LOG_DEBUG("Multicast consumer subscription name: %s\n", this->multicast_subscription_name.c_str());
+    ACA_LOG_DEBUG("ACA_PULSAR_MQ: Broker list -> %s\n", this->brokers_list.c_str());
+    ACA_LOG_DEBUG("ACA_PULSAR_MQ: Unicast consumer topic name -> %s\n", this->unicast_topic_name[0].c_str()); 
+    ACA_LOG_DEBUG("ACA_PULSAR_MQ: Unicast consumer subscription name -> %s\n", this->unicast_subscription_name.c_str());
+    ACA_LOG_DEBUG("ACA_PULSAR_MQ: Multicast consumer topic name -> %s\n", this->multicast_topic_name.c_str());
+    ACA_LOG_DEBUG("ACA_PULSAR_MQ: Multicast consumer subscription name -> %s\n", this->multicast_subscription_name.c_str());
 
     // Create the clients
     this->ptr_multicast_client= new Client(brokers);
@@ -112,23 +112,18 @@ string ACA_Message_Pulsar_Consumer::getMulticastSubscriptionName() const
 
 string ACA_Message_Pulsar_Consumer::getUnicastTopicName() const
 {
-  return this->unicast_topic_name;
+  string topic = "";
+  for (auto t: this->unicast_topic_name)
+    topic += (t+" ");
+  return topic;
 }
 
 string ACA_Message_Pulsar_Consumer::getUnicastSubscriptionName() const
 {
   return this->unicast_subscription_name;
 }
 
-//TODO: get recovered topic from database?
-string ACA_Message_Pulsar_Consumer::getRecoveredTopicName()
-{
-    return "recovered topic test";
-}
-
 bool ACA_Message_Pulsar_Consumer::unicastConsumerDispatched(int stickyHash){
-  Result result;
-  Consumer consumer;
   KeySharedPolicy keySharedPolicy;
 
   keySharedPolicy.setKeySharedMode(STICKY);
@@ -139,10 +134,9 @@ bool ACA_Message_Pulsar_Consumer::unicastConsumerDispatched(int stickyHash){
 
   //Use key shared mode
   this->unicast_consumer_config.setConsumerType(ConsumerKeyShared).setKeySharedPolicy(keySharedPolicy).setMessageListener(listener);
-  ACA_LOG_INFO("%s\n",this->unicast_topic_name.c_str());
-  result = this->ptr_unicast_client->subscribe(this->unicast_topic_name,this->unicast_subscription_name,this->unicast_consumer_config,this->unicast_consumer);
+  Result result = this->ptr_unicast_client->subscribe(this->unicast_topic_name,this->unicast_subscription_name,this->unicast_consumer_config,this->unicast_consumer);
   if (result != Result::ResultOk){
-    ACA_LOG_ERROR("Failed to subscribe unicast topic: %s\n", this->unicast_topic_name.c_str());
+    ACA_LOG_ERROR("ACA_PULSAR_MQ: ERROR, failed to subscribe unicast topic -> %s\n", this->getUnicastTopicName().c_str());
     return EXIT_FAILURE;
   }
 
@@ -156,54 +150,41 @@ bool ACA_Message_Pulsar_Consumer::multicastConsumerDispatched(){
   this->multicast_consumer_config.setMessageListener(listener);
   result = this->ptr_multicast_client->subscribe(this->multicast_topic_name,this->multicast_subscription_name,this->multicast_consumer_config,this->multicast_consumer);
   if (result != Result::ResultOk){
-    ACA_LOG_ERROR("Failed to subscribe multicast topic: %s\n", this->multicast_topic_name.c_str());
+    ACA_LOG_ERROR("ACA_PULSAR_MQ: ERROR, failed to subscribe multicast topic -> %s\n", this->multicast_topic_name.c_str());
     return EXIT_FAILURE;
   }
   return EXIT_SUCCESS;
 }
 
-bool ACA_Message_Pulsar_Consumer::unicastUnsubcribe()
+bool ACA_Message_Pulsar_Consumer::unicastUnsubscribeAll()  
 {
-    Result result;
-    if(this->unicast_topic_name==empty_topic){
-        ACA_LOG_INFO("The consumer already unsubscribe the unicast topic.\n");
+    if(this->unicast_topic_name.empty()){
+        ACA_LOG_INFO("ACA_PULSAR_MQ: Successfully to unsubscribe all the unicast topics.\n");
         return EXIT_SUCCESS;
     }
 
-    result=this->unicast_consumer.unsubscribe();
-    if (result != Result::ResultOk){
-        ACA_LOG_ERROR("Failed to unsubscribe unicast topic: %s\n", this->unicast_topic_name.c_str());
+    if (this->unicast_consumer.unsubscribe() == Result::ResultOk){
+        ACA_LOG_INFO("ACA_PULSAR_MQ: Successfully to unsubscribe all the unicast topics.\n");
+        this->unicast_topic_name.clear();
+        return EXIT_SUCCESS;
+    }
+    else{
+        ACA_LOG_ERROR("ACA_PULSAR_MQ: ERROR, failed to unsubscribe unicast topics -> %s.\n", this->getUnicastTopicName().c_str());
         return EXIT_FAILURE;
     }
-    this->unicast_topic_name=empty_topic;
-    return EXIT_SUCCESS;
 }
 
-bool ACA_Message_Pulsar_Consumer::unicastResubscribe(string topic, int stickyHash)
+bool ACA_Message_Pulsar_Consumer::unicastResubscribe(bool unSubscribe, string topic, string stickHash)
 {
-    bool result;
-
-    result = unicastUnsubcribe();
-
-    if (result==EXIT_SUCCESS){
+    if(!unSubscribe){
+      if(this->unicast_consumer.unsubscribe() == Result::ResultOk){ // this unsubscribes topics in pulsar, but doesn't clean the topic list of Consumer.
         setUnicastTopicName(topic);
-        result = unicastConsumerDispatched(stickyHash);
-        if (result==EXIT_SUCCESS) {
-            return EXIT_SUCCESS;
-        }
-    }
-    ACA_LOG_ERROR("Failed to resubscribe unicast topic: %s\n", topic.c_str());
-    return EXIT_FAILURE;
-}
-
-
-bool ACA_Message_Pulsar_Consumer::unicastResubscribe(bool isUnSubscribe, string topic, string stickHash)
-{
-    if(isUnSubscribe){
-        return unicastUnsubcribe();
+        return unicastConsumerDispatched(stoi(stickHash));
+      }
+      return EXIT_FAILURE;
     }
     else{
-        return unicastResubscribe(topic, std::stoi(stickHash));
+        return unicastUnsubscribeAll();
     }
 }
 
@@ -224,7 +205,7 @@ void ACA_Message_Pulsar_Consumer::setMulticastSubscriptionName(string subscripti
 
 void ACA_Message_Pulsar_Consumer::setUnicastTopicName(string topic)
 {
-  this->unicast_topic_name = topic;
+  this->unicast_topic_name.push_back(topic);
 }
 
 void ACA_Message_Pulsar_Consumer::setUnicastSubscriptionName(string subscription_name)

src/comm/aca_message_pulsar_producer.cpp

@@ -55,60 +55,56 @@ void ACA_Message_Pulsar_Producer::setTopicName(string topic)
 
 bool ACA_Message_Pulsar_Producer::publish(string message)
 {
-  Result result;
-
   // Create a producer
   Producer producer;
-  result = this->ptr_client->createProducer(this->topic_name,producer);
-  if(result != ResultOk){
-    ACA_LOG_ERROR("Failed to create producer, result=%d.\n", result);
+  Result createProducerResult = this->ptr_client->createProducer(this->topic_name,producer);
+  if(createProducerResult != ResultOk){
+    ACA_LOG_ERROR("ACA_PULSAR_MQ: Failed to create producer, result=%d.\n", createProducerResult);
     return EXIT_FAILURE;
   }
 
   // Create a message
   Message msg = MessageBuilder().setContent(message).build();
-  result = producer.send(msg);
-  if(result != ResultOk){
-    ACA_LOG_ERROR("Failed to send message %s.\n", message.c_str());
+
+  if(producer.send(msg) != ResultOk){
+    ACA_LOG_ERROR("ACA_PULSAR_MQ: Failed to send message -> %s.\n", message.c_str());
     return EXIT_FAILURE;
   }
+  else{
+    ACA_LOG_INFO("ACA_PULSAR_MQ: Successfully send message %s.\n", message.c_str());
 
-  ACA_LOG_INFO("Successfully send message %s\n", message.c_str());
-
-  // Flush all produced messages
-  producer.flush();
-  producer.close();
-  return EXIT_SUCCESS;
-
+    // Flush all produced messages
+    producer.flush();
+    producer.close();
+    return EXIT_SUCCESS;
+  }
 }
 
 bool ACA_Message_Pulsar_Producer::publish(string message, string orderingKey)
 {
-    Result result;
-
     // Create a producer
     Producer producer;
-    result = this->ptr_client->createProducer(this->topic_name,producer);
-    if(result != ResultOk){
-        ACA_LOG_ERROR("Failed to create producer, result=%d.\n", result);
+    Result createProducerResult = this->ptr_client->createProducer(this->topic_name,producer);
+    if(createProducerResult != ResultOk){
+        ACA_LOG_ERROR("ACA_PULSAR_MQ: Failed to create producer, result=%d.\n", createProducerResult);
         return EXIT_FAILURE;
     }
 
     // Create a message
     Message msg = MessageBuilder().setContent(message).setOrderingKey(orderingKey).build();
-    result = producer.send(msg);
-    if(result != ResultOk){
-        ACA_LOG_ERROR("Failed to send message %s.\n", message.c_str());
+
+    if(producer.send(msg) != ResultOk){
+        ACA_LOG_ERROR("ACA_PULSAR_MQ: Failed to send message %s.\n", message.c_str());
         return EXIT_FAILURE;
     }
+    else{
+        ACA_LOG_INFO("ACA_PULSAR_MQ: Successfully send message %s.\n", message.c_str());
 
-    ACA_LOG_INFO("Successfully send message %s\n", message.c_str());
-
-    // Flush all produced messages
-    producer.flush();
-    producer.close();
-    return EXIT_SUCCESS;
-
+        // Flush all produced messages
+        producer.flush();
+        producer.close();
+        return EXIT_SUCCESS;
+    }
 }
 void ACA_Message_Pulsar_Producer::setBrokers(string brokers)
 {

test/func_tests/gs_tests.cpp

@@ -58,6 +58,8 @@ string g_ncm_port = EMPTY_STRING;
 string g_grpc_server_port = EMPTY_STRING;
 // by default, this should run as GRCP client, unless specified by the corresponding flag.
 bool g_run_as_server = false;
+bool g_run_as_topic_client = false;
+
 GoalStateProvisionerAsyncServer *g_grpc_server = NULL;
 GoalStateProvisionerClientImpl *g_grpc_client = NULL;
 // GoalStateProvisionerServer *g_test_grcp_server = NULL;
@@ -812,7 +814,7 @@ int main(int argc, char *argv[])
   signal(SIGINT, aca_signal_handler);
   signal(SIGTERM, aca_signal_handler);
 
-  while ((option = getopt(argc, argv, "s:p:dm")) != -1) {
+  while ((option = getopt(argc, argv, "s:p:dmt")) != -1) {
     switch (option) {
     case 's':
       g_grpc_server_ip = optarg;
@@ -826,14 +828,18 @@ int main(int argc, char *argv[])
     case 'm':
       g_run_as_server = true;
       break;
+    case 't':
+      g_run_as_topic_client = true;
+      break;
     default: /* the '?' case when the option is not recognized */
       /* specifying port not avaiable for now */
       fprintf(stderr,
               "Usage: %s\n"
               "\t\t[-s grpc server]\n"
               "\t\t[-p grpc port]\n"
               "\t\t[-d enable debug mode]\n"
-              "\t\t[-m If this flag is passed in, gs test runs as grpc server, which listens on localhost:54321; otherwise it runs as a grpc client]\n",
+              "\t\t[-m If this flag is passed in, gs test runs as grpc server, which listens on localhost:54321;]\n",
+              "\t\t[-t If this flag is passed in, gs test runs as grpc topic client, which publishes topic subscribe requests to localhost:50002; otherwise it runs as a grpc client]\n",
               argv[0]);
       exit(EXIT_FAILURE);
     }
@@ -853,9 +859,12 @@ int main(int argc, char *argv[])
 
   if (g_run_as_server) {
     rc = RunServer();
-  } else {
-//    rc = run_as_client();
-    rc = run_as_topic_client();
+  }
+  else if (g_run_as_topic_client){
+      rc = run_as_topic_client();
+  }
+  else {
+      rc = run_as_client();
   }
   // Verify that the version of the library that we linked against is
   // compatible with the version of the headers we compiled against.