vector<TimerEvent*> Timer::GetExpiredTimerEvent(const Timestamp& now_ts)
{
    LOG_VERBOSE << "us:" << now_ts.GetTimeUs() << endl;
    vector<TimerEvent*> expired_events;

    PrintWaitEvent();

    auto iter = wait_event_.lower_bound(now_ts.GetTimeUs());
    auto it = wait_event_.begin();
    while (it != iter)
    {
        auto timer_it = id_timer_map_.find(it->second);
        if (timer_it != id_timer_map_.end())
        {
            expired_events.push_back(timer_it->second.get());
        }
        else
        {
            LOG_VERBOSE << "timer id:" << it->second << " has been canceled" << endl;
        }

        it = wait_event_.erase(it);

        LOG_VERBOSE << "some timer event happend" << endl;
    }

    return expired_events;
}

DateTime::DateTime()
{
	Timestamp now;
	_utcTime = now.utcTime();
	computeGregorian(julianDay());
	computeDaytime();
}

void TimerTest::testScheduleInterval()
{
	Timer timer;

	Timestamp time;

	TimerTask::Ptr pTask = new TimerTaskAdapter<TimerTest>(*this, &TimerTest::onTimer);

	assert (pTask->lastExecution() == 0);

	timer.schedule(pTask, 500, 500);

	_event.wait();
	assert (time.elapsed() >= 590000);
	assert (pTask->lastExecution().elapsed() < 130000);

	_event.wait();
	assert (time.elapsed() >= 1190000);
	assert (pTask->lastExecution().elapsed() < 130000);

	_event.wait();
	assert (time.elapsed() >= 1790000);
	assert (pTask->lastExecution().elapsed() < 130000);

	pTask->cancel();
	assert (pTask->isCancelled());
}

void LiveSLAMWrapper::newImageCallback(const cv::Mat& img, Timestamp imgTime)
{
	++ imageSeqNumber;

	// Convert image to grayscale, if necessary
	cv::Mat grayImg;
	if (img.channels() == 1)
		grayImg = img;
	else
		cvtColor(img, grayImg, CV_RGB2GRAY);
	

	// Assert that we work with 8 bit images
	assert(grayImg.elemSize() == 1);
	assert(fx != 0 || fy != 0);


	// need to initialize
	if(!isInitialized)
	{
		monoOdometry->randomInit(grayImg.data, img.data, imgTime.toSec(), 1);
		isInitialized = true;
	}
	else if(isInitialized && monoOdometry != nullptr)
	{
		monoOdometry->trackFrame(grayImg.data, img.data, imageSeqNumber,false,imgTime.toSec());
	}
}

StatusWith<OplogFetcher::DocumentsInfo> OplogFetcher::validateDocuments(
    const Fetcher::Documents& documents, bool first, Timestamp lastTS) {
    if (first && documents.empty()) {
        return Status(ErrorCodes::OplogStartMissing,
                      str::stream() << "The first batch of oplog entries is empty, but expected at "
                                       "least 1 document matching ts: "
                                    << lastTS.toString());
    }

    DocumentsInfo info;
    // The count of the bytes of the documents read off the network.
    info.networkDocumentBytes = 0;
    info.networkDocumentCount = 0;
    for (auto&& doc : documents) {
        info.networkDocumentBytes += doc.objsize();
        ++info.networkDocumentCount;

        // If this is the first response (to the $gte query) then we already applied the first doc.
        if (first && info.networkDocumentCount == 1U) {
            continue;
        }

        // Check to see if the oplog entry goes back in time for this document.
        const auto docOpTime = OpTime::parseFromOplogEntry(doc);
        // entries must have a "ts" field.
        if (!docOpTime.isOK()) {
            return docOpTime.getStatus();
        }

        info.lastDocument = {doc["h"].numberLong(), docOpTime.getValue()};

        const auto docTS = info.lastDocument.opTime.getTimestamp();
        if (lastTS >= docTS) {
            return Status(ErrorCodes::OplogOutOfOrder,
                          str::stream() << "Out of order entries in oplog. lastTS: "
                                        << lastTS.toString()
                                        << " outOfOrderTS:"
                                        << docTS.toString()
                                        << " in batch with "
                                        << info.networkDocumentCount
                                        << "docs; first-batch:"
                                        << first
                                        << ", doc:"
                                        << doc);
        }
        lastTS = docTS;
    }

    // These numbers are for the documents we will apply.
    info.toApplyDocumentCount = documents.size();
    info.toApplyDocumentBytes = info.networkDocumentBytes;
    if (first) {
        // The count is one less since the first document found was already applied ($gte $ts query)
        // and we will not apply it again.
        --info.toApplyDocumentCount;
        auto alreadyAppliedDocument = documents.cbegin();
        info.toApplyDocumentBytes -= alreadyAppliedDocument->objsize();
    }
    return info;
}

StatusWith<RollBackLocalOperations::RollbackCommonPoint> syncRollBackLocalOperations(
    const OplogInterface& localOplog,
    const OplogInterface& remoteOplog,
    const RollBackLocalOperations::RollbackOperationFn& rollbackOperation) {
    auto remoteIterator = remoteOplog.makeIterator();
    auto remoteResult = remoteIterator->next();
    if (!remoteResult.isOK()) {
        return StatusWith<RollBackLocalOperations::RollbackCommonPoint>(
            ErrorCodes::InvalidSyncSource, "remote oplog empty or unreadable");
    }

    RollBackLocalOperations finder(localOplog, rollbackOperation);
    Timestamp theirTime;
    while (remoteResult.isOK()) {
        theirTime = remoteResult.getValue().first["ts"].timestamp();
        BSONObj theirObj = remoteResult.getValue().first;
        auto result = finder.onRemoteOperation(theirObj);
        if (result.isOK()) {
            return result.getValue();
        } else if (result.getStatus().code() != ErrorCodes::NoSuchKey) {
            return result;
        }
        remoteResult = remoteIterator->next();
    }

    severe() << "rollback error RS100 reached beginning of remote oplog";
    log() << "  them:      " << remoteOplog.toString();
    log() << "  theirTime: " << theirTime.toStringLong();
    return StatusWith<RollBackLocalOperations::RollbackCommonPoint>(
        ErrorCodes::NoMatchingDocument, "RS100 reached beginning of remote oplog [1]");
}

void OplogBufferCollection::pushAllNonBlocking(OperationContext* txn,
        Batch::const_iterator begin,
        Batch::const_iterator end) {
    if (begin == end) {
        return;
    }
    size_t numDocs = std::distance(begin, end);
    Batch docsToInsert(numDocs);
    Timestamp ts;
    std::transform(begin, end, docsToInsert.begin(), [&ts](const Value& value) {
        auto pair = addIdToDocument(value);
        invariant(ts.isNull() || pair.second > ts);
        ts = pair.second;
        return pair.first;
    });

    stdx::lock_guard<stdx::mutex> lk(_mutex);
    auto status = _storageInterface->insertDocuments(txn, _nss, docsToInsert);
    fassertStatusOK(40161, status);

    _lastPushedTimestamp = ts;
    _count += numDocs;
    _size += std::accumulate(begin, end, 0U, [](const size_t& docSize, const Value& value) {
        return docSize + size_t(value.objsize());
    });
    _cvNoLongerEmpty.notify_all();
}

void TimerQueue::reset(std::vector<TimerQueue::Entry> expired, Timestamp  now)
{
	for (auto iter : expired)
	{
		ActiveTimer timer(iter.second, iter.second->sequeue());
		if (iter.second->repeat() && 
				cancleTimers_.find(timer) == cancleTimers_.end())
		{
			iter.second->restart(now);
			//timers_.insert(iter);
			//activeTimers_.insert(timer);
			insert(iter.second);
		}
		else
		{
			delete iter.second;
		}
	}

	Timestamp time;
	if (!timers_.empty())
	{
		time = timers_.begin()->second->expiration();
	}

	if (time.valid())
	{
		ext::resetTimerfd(timerfd_, time);
	}
}

void TimerQueue::reset(const std::vector<Entry>& expired, Timestamp now)
{
    Timestamp nextExpire;

    for (std::vector<Entry>::const_iterator it = expired.begin();
        it != expired.end(); it++)
    {
        ActiveTimer timer(it->second, it->second->sequence());
        if (it->second->repeat()
		&& cancelingTimers_.find(timer) == cancelingTimers_.end())
        {
            it->second->restart(now);
            insert(it->second);
        }
        else
        {
            delete it->second;
        }
    }

    if (!timers_.empty())
    {
        nextExpire = timers_.begin()->second->expiration();
    }
    
    if (nextExpire.valid())
    {
        resetTimerfd(timerfd_, nextExpire);
    }
}

void SyncTail::handleSlaveDelay(const BSONObj& lastOp) {
    ReplicationCoordinator* replCoord = getGlobalReplicationCoordinator();
    int slaveDelaySecs = durationCount<Seconds>(replCoord->getSlaveDelaySecs());

    // ignore slaveDelay if the box is still initializing. once
    // it becomes secondary we can worry about it.
    if (slaveDelaySecs > 0 && replCoord->getMemberState().secondary()) {
        const Timestamp ts = lastOp["ts"].timestamp();
        long long a = ts.getSecs();
        long long b = time(0);
        long long lag = b - a;
        long long sleeptime = slaveDelaySecs - lag;
        if (sleeptime > 0) {
            uassert(12000,
                    "rs slaveDelay differential too big check clocks and systems",
                    sleeptime < 0x40000000);
            if (sleeptime < 60) {
                sleepsecs((int)sleeptime);
            } else {
                warning() << "slavedelay causing a long sleep of " << sleeptime << " seconds";
                // sleep(hours) would prevent reconfigs from taking effect & such!
                long long waitUntil = b + sleeptime;
                while (time(0) < waitUntil) {
                    sleepsecs(6);

                    // Handle reconfigs that changed the slave delay
                    if (durationCount<Seconds>(replCoord->getSlaveDelaySecs()) != slaveDelaySecs)
                        break;
                }
            }
        }
    }  // endif slaveDelay
}

int Timestamp::Compare(const Timestamp &aCompare) const
{
    uint64_t thisSeconds = GetSeconds();
    uint64_t compareSeconds = aCompare.GetSeconds();
    uint16_t thisTicks = GetTicks();
    uint16_t compareTicks = aCompare.GetTicks();
    int rval;

    if (compareSeconds > thisSeconds)
    {
        rval = 1;
    }
    else if (compareSeconds < thisSeconds)
    {
        rval = -1;
    }
    else if (compareTicks > thisTicks)
    {
        rval = 1;
    }
    else if (compareTicks < thisTicks)
    {
        rval = -1;
    }
    else
    {
        rval = 0;
    }

    return rval;
}

void sslTest2()
{
	tracef("ssl test 2 begin: ssl send data.");
	Context::Ptr pContext = new Context(Context::TLSV1_CLIENT_USE, "", Context::VERIFY_NONE);
	SecureStreamSocket sss(pContext);
	SocketAddress sa("127.0.0.1", 12222);
	sss.connect(sa, Timespan(3, 0));
	DynamicStruct ds;
	ds["type"] = "request";
	ds["action"] = "server.token";
	Timestamp t;
	UInt64 tms = t.epochMicroseconds();
	char tms_str[32];
	snprintf(tms_str, 31, "%llu", tms);
	std::string key = "alpha2015";
	key += tms_str;
	MD5Engine md5;
	md5.update(key);
	const DigestEngine::Digest& digest = md5.digest();
	std::string md5key = DigestEngine::digestToHex(digest);
	DynamicStruct param;
	param["key"] = md5key;
	param["timestamp"] = tms_str;
	param["dev_name"] = "lock1";
	param["dev_type"] = "sc-01";
	param["uuid"] = "SC00000001";
	ds["param"] = param;

	tracef("data send: %s.", ds.toString().c_str());
	sss.sendBytes(ds.toString().c_str(), ds.toString().length());
	sss.close();
	tracef("socket closed.");
	tracef("ssl test 2 finished.\n");
}

void SSLContext::flushSessionCache() 
{
    assert(_usage == SERVER_USE);

    Timestamp now;
    SSL_CTX_flush_sessions(_sslContext, static_cast<long>(now.epochTime()));
}

void ReqGetNewMsgs::OnSuccess()
{
  PXml p = m_xml.getChildNode(0);
  Assert(SafeStrCmp(p.getName(), "now"));
  timestamp = p.getText();

  p = m_xml.getChildNode(1);
  if (SafeStrCmp(p.getName(), "msgs"))
  {
    int numMsgs = p.nChildNode();
    for (int i = 0; i < numMsgs; i++)
    {
      PXml msg = p.getChildNode(i);

      int id = atoi(msg.getChildNode(0).getText());
      int senderId = atoi(msg.getChildNode(1).getText());
      timestamp = msg.getChildNode(2).getText();
      Timestamp whenSent = atoi(timestamp.c_str());
      std::string text = msg.getChildNode(3).getText();
      text = DecodeMsg(text); ////Replace(text, "_", " "); // replace underscores with spaces -- TODO punctuation 
      int recipId = atoi(msg.getChildNode(4).getText());

std::cout << "GOT NEW MSG!! ID: " << id << " sender: " << senderId << " sent: " << whenSent.ToString() << " text: " << text << "\n";

      TheMsgManager::Instance()->QueueMsg(MsgManager::Msg(id, text, senderId, recipId, whenSent));
    }
  }
  else
  {
    ShowError("Unexpected format for msgs response");
  }
}

Ref<Database> PackageService::buildLookupDB(const String& path, StreamSourceMap& packCfgs)
{
	Ref<Database> db = Database::open(path);

	db->exec("DROP TABLE IF EXISTS packs; CREATE TABLE packs (name PRIMARY KEY, timestamp INT)");
	db->exec("DROP TABLE IF EXISTS lookup; CREATE TABLE lookup (type, key, subtype, pack, entry, PRIMARY KEY (type, key))");
	db->exec("DROP INDEX IF EXISTS lookup_packidx; CREATE INDEX lookup_packidx ON lookup (pack ASC)");

	Ref<Database::Query> packInsert = 
		db->prepare("INSERT INTO packs (name, timestamp) VALUES (?, ?)");

	Ref<Database::Query> lookupInsert = 
		db->prepare("INSERT INTO lookup (type, key, subtype, pack, entry) VALUES ($type, $key, $subtype, $pack, $entry)");

	for (StreamSourceMap::iterator itr = packCfgs.begin(), end = packCfgs.end(); itr != end; ++itr)
	{
		const String& packName = itr->first;
		Ref<Settings> cfg = Settings::load(itr->second);

		if (cfg == NULL)
		{
			LOG(0, "*** can't load '%s'\n", cfg->getUrl().c_str());
			continue;
		}

		Timestamp time = itr->second->getTimestamp();

		packInsert->reset();
		packInsert->bind(1, packName);
		packInsert->bind(2, time.getUnixTime64());

		packInsert->step();

		Ref<Settings> section = cfg->getSection("lookup");
		if (section == NULL)
			continue;

		for (Settings::Iterator itr = section->begin(), end = section->end(); itr != end; ++itr)
		{
			String type = itr->first;
			String key = itr->second;

			String sub, entry;

			splitLookupEntry(type, key, sub, entry);

			lookupInsert->reset();
			lookupInsert->bind("$type", type);
			lookupInsert->bind("$key", key);
			lookupInsert->bind("$subtype", sub);
			lookupInsert->bind("$entry", entry);
			lookupInsert->bind("$pack", packName);

			lookupInsert->step();
		}
	}

	return db;
}

auto greater::operator()<Timestamp&, Timestamp&>(Timestamp& a, Timestamp& b) const noexcept
{
    if (a.is_null())
        return false;
    if (b.is_null())
        return true;
    return a > b;
}

int main(int argc, char const *argv[])
{
	Timestamp now = Timestamp::now();

	cout << now.toString() << endl;
	cout << now.toFormatString() << endl;
	return 0;
}

 /**
  * This function returns the fraction of time elapsed after the beginning 
  * till the end
  */
 double fraction(Timestamp currentTime, Timestamp endTime) {
     if ( (currentTime.after(*this)) && (endTime.after(currentTime)) ) {
         return(((double)currentTime.subUsec(*this)) /
                ((double)endTime.subUsec(*this)));
     } else {
         return -1.0;
     }
 }

std::pair<BSONObj, Timestamp> OplogBufferCollection::addIdToDocument(const BSONObj& orig) {
    invariant(!orig.isEmpty());
    BSONObjBuilder bob;
    Timestamp ts = orig["ts"].timestamp();
    invariant(!ts.isNull());
    bob.append("_id", ts);
    bob.append(kOplogEntryFieldName, orig);
    return std::pair<BSONObj, Timestamp> {bob.obj(), ts};
}

void delay_loop( unsigned long usec ) {
    Timestamp end;
    end.add( usec * 1e-6 );

    Timestamp now;
    while ( now.before( end ) ) {
        now.setnow();
    }
}