void Monitor::dispatch_op(MonOpRequestRef op)
{
  op->mark_event("mon:dispatch_op");
  MonSession *s = op->get_session();
  assert(s);
  if (s->closed) {
    dout(10) << " session closed, dropping " << op->get_req() << dendl;
    return;
  }

  /* we will consider the default type as being 'monitor' until proven wrong */
  op->set_type_monitor();
  /* deal with all messages that do not necessarily need caps */
  bool dealt_with = true;
  switch (op->get_req()->get_type()) {
    // auth
    case MSG_MON_GLOBAL_ID:
    case CEPH_MSG_AUTH:
      op->set_type_service();
      /* no need to check caps here */
      paxos_service[PAXOS_AUTH]->dispatch(op);
      break;

    case CEPH_MSG_PING:
      handle_ping(op);
      break;

    /* MMonGetMap may be used by clients to obtain a monmap *before*
     * authenticating with the monitor.  We need to handle these without
     * checking caps because, even on a cluster without cephx, we only set
     * session caps *after* the auth handshake.  A good example of this
     * is when a client calls MonClient::get_monmap_privately(), which does
     * not authenticate when obtaining a monmap.
     */
    case CEPH_MSG_MON_GET_MAP:
      handle_mon_get_map(op);
      break;

    case CEPH_MSG_MON_METADATA:
      return handle_mon_metadata(op);

    default:
      dealt_with = false;
      break;
  }
  if (dealt_with)
    return;

  /* well, maybe the op belongs to a service... */
  op->set_type_service();
  /* deal with all messages which caps should be checked somewhere else */
  dealt_with = true;
  switch (op->get_req()->get_type()) {

    // OSDs
    case CEPH_MSG_MON_GET_OSDMAP:
    case CEPH_MSG_POOLOP:
    case MSG_OSD_BEACON:
    case MSG_OSD_MARK_ME_DOWN:
    case MSG_OSD_FULL:
    case MSG_OSD_FAILURE:
    case MSG_OSD_BOOT:
    case MSG_OSD_ALIVE:
    case MSG_OSD_PGTEMP:
    case MSG_OSD_PG_CREATED:
    case MSG_REMOVE_SNAPS:
      paxos_service[PAXOS_OSDMAP]->dispatch(op);
      break;

    // MDSs
    case MSG_MDS_BEACON:
    case MSG_MDS_OFFLOAD_TARGETS:
      paxos_service[PAXOS_MDSMAP]->dispatch(op);
      break;

    // Mgrs
    case MSG_MGR_BEACON:
      paxos_service[PAXOS_MGR]->dispatch(op);
      break;

    // MgrStat
    case CEPH_MSG_STATFS:
      // this is an ugly hack, sorry!  force the version to 1 so that we do
      // not run afoul of the is_readable() paxos check.  the client is going
      // by the pgmonitor version and the MgrStatMonitor version will lag behind
      // that until we complete the upgrade.  The paxos ordering crap really
      // doesn't matter for statfs results, so just kludge around it here.
      if (osdmon()->osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
	((MStatfs*)op->get_req())->version = 1;
      }
    case MSG_MON_MGR_REPORT:
    case MSG_GETPOOLSTATS:
      paxos_service[PAXOS_MGRSTAT]->dispatch(op);
      break;

    // pg
    case MSG_PGSTATS:
      paxos_service[PAXOS_PGMAP]->dispatch(op);
      break;

    // log
    case MSG_LOG:
      paxos_service[PAXOS_LOG]->dispatch(op);
      break;

    // handle_command() does its own caps checking
    case MSG_MON_COMMAND:
      op->set_type_command();
      handle_command(op);
      break;

    default:
      dealt_with = false;
      break;
  }
  if (dealt_with)
    return;

  /* nop, looks like it's not a service message; revert back to monitor */
  op->set_type_monitor();

  /* messages we, the Monitor class, need to deal with
   * but may be sent by clients. */

  if (!op->get_session()->is_capable("mon", MON_CAP_R)) {
    dout(5) << __func__ << " " << op->get_req()->get_source_inst()
            << " not enough caps for " << *(op->get_req()) << " -- dropping"
            << dendl;
    goto drop;
  }

  dealt_with = true;
  switch (op->get_req()->get_type()) {

    // misc
    case CEPH_MSG_MON_GET_VERSION:
      handle_get_version(op);
      break;

    case CEPH_MSG_MON_SUBSCRIBE:
      /* FIXME: check what's being subscribed, filter accordingly */
      handle_subscribe(op);
      break;

    default:
      dealt_with = false;
      break;
  }
  if (dealt_with)
    return;

  if (!op->is_src_mon()) {
    dout(1) << __func__ << " unexpected monitor message from"
            << " non-monitor entity " << op->get_req()->get_source_inst()
            << " " << *(op->get_req()) << " -- dropping" << dendl;
    goto drop;
  }

  /* messages that should only be sent by another monitor */
  dealt_with = true;
  switch (op->get_req()->get_type()) {

    case MSG_ROUTE:
      handle_route(op);
      break;

    case MSG_MON_PROBE:
      handle_probe(op);
      break;

    // Sync (i.e., the new slurp, but on steroids)
    case MSG_MON_SYNC:
      handle_sync(op);
      break;
    case MSG_MON_SCRUB:
      handle_scrub(op);
      break;

    /* log acks are sent from a monitor we sent the MLog to, and are
       never sent by clients to us. */
    case MSG_LOGACK:
      log_client.handle_log_ack((MLogAck*)op->get_req());
      break;

    // monmap
    case MSG_MON_JOIN:
      op->set_type_service();
      paxos_service[PAXOS_MONMAP]->dispatch(op);
      break;

    // paxos
    case MSG_MON_PAXOS:
      {
        op->set_type_paxos();
        MMonPaxos *pm = static_cast<MMonPaxos*>(op->get_req());
        if (!op->get_session()->is_capable("mon", MON_CAP_X)) {
          //can't send these!
          break;
        }

        if (state == STATE_SYNCHRONIZING) {
          // we are synchronizing. These messages would do us no
          // good, thus just drop them and ignore them.
          dout(10) << __func__ << " ignore paxos msg from "
            << pm->get_source_inst() << dendl;
          break;
        }

        // sanitize
        if (pm->epoch > get_epoch()) {
          bootstrap();
          break;
        }
        if (pm->epoch != get_epoch()) {
          break;
        }

        paxos->dispatch(op);
      }
      break;

    // elector messages
    case MSG_MON_ELECTION:
      op->set_type_election();
      //check privileges here for simplicity
      if (!op->get_session()->is_capable("mon", MON_CAP_X)) {
        dout(0) << "MMonElection received from entity without enough caps!"
          << op->get_session()->caps << dendl;
        break;
      }
      if (!is_probing() && !is_synchronizing()) {
        elector.dispatch(op);
      }
      break;

    case MSG_FORWARD:
      handle_forward(op);
      break;

    case MSG_TIMECHECK:
      handle_timecheck(op);
      break;

    case MSG_MON_HEALTH:
      health_monitor->dispatch(op);
      break;

    case MSG_MON_HEALTH_CHECKS:
      op->set_type_service();
      paxos_service[PAXOS_HEALTH]->dispatch(op);
      break;

    default:
      dealt_with = false;
      break;
  }
  if (!dealt_with) {
    dout(1) << "dropping unexpected " << *(op->get_req()) << dendl;
    goto drop;
  }
  return;

drop:
  return;
}