logclient: not a dispatcher

Let MonClient and Monitor handle delivery of messages.  This puts them in
control and lets them trigger sending of more messages when we have a
bunch queued.

Signed-off-by: Sage Weil <sage@inktank.com>

3 files changed, 24 insertions, 8 deletions

diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
index b002e780b01..4efaf6503a8 100644
--- a/src/mon/MonClient.cc
+++ b/src/mon/MonClient.cc
@@ -169,6 +169,7 @@ bool MonClient::ms_dispatch(Message *m)
   case CEPH_MSG_AUTH_REPLY:
   case CEPH_MSG_MON_SUBSCRIBE_ACK:
   case CEPH_MSG_MON_GET_VERSION_REPLY:
+  case MSG_LOGACK:
     break;
   default:
     return false;
@@ -195,10 +196,25 @@ bool MonClient::ms_dispatch(Message *m)
     break;
   case CEPH_MSG_MON_GET_VERSION_REPLY:
     handle_get_version_reply((MMonGetVersionReply*)m);
+    break;
+  case MSG_LOGACK:
+    if (log_client) {
+      log_client->handle_log_ack((MLogAck*)m);
+    } else {
+      m->put();
+    }
+    break;
   }
   return true;
 }
 
+void MonClient::send_log()
+{
+  Message *lm = log_client->get_mon_log_message();
+  if (lm)
+    _send_mon_message(lm);
+}
+
 void MonClient::handle_monmap(MMonMap *m)
 {
   ldout(cct, 10) << "handle_monmap " << *m << dendl;
@@ -368,9 +384,7 @@ void MonClient::handle_auth(MAuthReply *m)
 
       if (log_client) {
 	log_client->reset_session();
-	Message *lm = log_client->get_mon_log_message();
-	if (lm)
-	  _send_mon_message(lm);
+	send_log();
       }
     }
   
@@ -510,9 +524,7 @@ void MonClient::tick()
    
     if (state == MC_STATE_HAVE_SESSION &&
 	log_client) {
-      Message *m = log_client->get_mon_log_message();
-      if (m)
-	_send_mon_message(m);
+      send_log();
     }
   }
 
diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h
index 8763950f4fb..884f77016e7 100644
--- a/src/mon/MonClient.h
+++ b/src/mon/MonClient.h
@@ -68,6 +68,7 @@ private:
   bool initialized;
 
   LogClient *log_client;
+  void send_log();
 
   AuthSupported *auth_supported;
 
diff --git a/src/mon/Monitor.cc b/src/mon/Monitor.cc
index 46c1236681a..3f2f287e08a 100644
--- a/src/mon/Monitor.cc
+++ b/src/mon/Monitor.cc
@@ -97,7 +97,7 @@ Monitor::Monitor(CephContext* cct_, string nm, MonitorStore *s, Messenger *m, Mo
   has_ever_joined(false),
   logger(NULL), cluster_logger(NULL), cluster_logger_registered(false),
   monmap(map),
-  clog(cct_, messenger, monmap, NULL, LogClient::FLAG_MON),
+  clog(cct_, messenger, monmap, LogClient::FLAG_MON),
   key_server(cct),
   auth_supported(cct),
   store(s),
@@ -356,7 +356,6 @@ int Monitor::init()
 
   // i'm ready!
   messenger->add_dispatcher_tail(this);
-  messenger->add_dispatcher_head(&clog);
   
   // start ticker
   timer.init();
@@ -1703,6 +1702,10 @@ bool Monitor::_ms_dispatch(Message *m)
       paxos_service[PAXOS_LOG]->dispatch((PaxosServiceMessage*)m);
       break;
 
+    case MSG_LOGACK:
+      clog.handle_log_ack((MLogAck*)m);
+      break;
+
       // monmap
     case MSG_MON_JOIN:
       paxos_service[PAXOS_MONMAP]->dispatch((PaxosServiceMessage*)m);