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;
};
}