replica_set_monitor.h

/*    Copyright 2014 MongoDB Inc.
 *
 *    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.
 */

#pragma once

#include <boost/shared_ptr.hpp>
#include <string>
#include <set>

#include "mongo/base/disallow_copying.h"
#include "mongo/base/string_data.h"
#include "mongo/stdx/functional.h"
#include "mongo/util/net/hostandport.h"

namespace mongo {
class BSONObj;
class ReplicaSetMonitor;
class TagSet;
struct ReadPreferenceSetting;
typedef boost::shared_ptr<ReplicaSetMonitor> ReplicaSetMonitorPtr;

class ReplicaSetMonitor {
    MONGO_DISALLOW_COPYING(ReplicaSetMonitor);

public:
    class Refresher;

    typedef stdx::function<void(const std::string& setName, const std::string& newConnectionString)>
        ConfigChangeHook;

    ReplicaSetMonitor(StringData name, const std::set<HostAndPort>& seeds);

    HostAndPort getHostOrRefresh(const ReadPreferenceSetting& criteria);

    HostAndPort getMasterOrUassert();

    Refresher startOrContinueRefresh();

    void failedHost(const HostAndPort& host);

    bool isPrimary(const HostAndPort& host) const;

    bool isHostUp(const HostAndPort& host) const;

    int getConsecutiveFailedScans() const;

    std::string getName() const;

    std::string getServerAddress() const;

    bool contains(const HostAndPort& server) const;

    void appendInfo(BSONObjBuilder& b) const;

    static void createIfNeeded(const std::string& name, const std::set<HostAndPort>& servers);

    static ReplicaSetMonitorPtr get(const std::string& name, bool createFromSeed = false);

    static std::set<std::string> getAllTrackedSets();

    static void remove(const std::string& name, bool clearSeedCache = false);

    static void setConfigChangeHook(ConfigChangeHook hook);

    static Status initialize();

    static Status shutdown(int gracePeriodMillis = 0);

    static int maxConsecutiveFailedChecks;

    //
    // internal types (defined in replica_set_monitor_internal.h)
    //

    struct IsMasterReply;
    struct ScanState;
    struct SetState;
    typedef boost::shared_ptr<ScanState> ScanStatePtr;
    typedef boost::shared_ptr<SetState> SetStatePtr;

    //
    // FOR TESTING ONLY
    //

    explicit ReplicaSetMonitor(const SetStatePtr& initialState) : _state(initialState) {}

    static bool useDeterministicHostSelection;

private:
    const SetStatePtr _state;  // never NULL
};

class ReplicaSetMonitor::Refresher {
public:
    HostAndPort refreshUntilMatches(const ReadPreferenceSetting& criteria) {
        return _refreshUntilMatches(&criteria);
    };

    void refreshAll() {
        _refreshUntilMatches(NULL);
    }

    //
    // Remaining methods are only for testing and internal use.
    // Callers are responsible for holding SetState::mutex before calling any of these methods.
    //

    explicit Refresher(const SetStatePtr& setState);

    struct NextStep {
        enum StepKind {
            CONTACT_HOST,  
            WAIT,          
            DONE,          
        };

        explicit NextStep(StepKind step, const HostAndPort& host = HostAndPort())
            : step(step), host(host) {}

        StepKind step;
        HostAndPort host;
    };

    NextStep getNextStep();

    void receivedIsMaster(const HostAndPort& from, int64_t latencyMicros, const BSONObj& reply);

    void failedHost(const HostAndPort& host);

    bool startedNewScan() const {
        return _startedNewScan;
    }

    static ScanStatePtr startNewScan(const SetState* set);

private:
    bool receivedIsMasterFromMaster(const IsMasterReply& reply);

    void receivedIsMasterBeforeFoundMaster(const IsMasterReply& reply);

    HostAndPort _refreshUntilMatches(const ReadPreferenceSetting* criteria);

    // Both pointers are never NULL
    SetStatePtr _set;
    ScanStatePtr _scan;  // May differ from _set->currentScan if a new scan has started.
    bool _startedNewScan;
};
}