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