SERVER-22774 Copied BackgroundSync::_fetcherCallback logic to OplogFetcher

1 files changed, 23 insertions, 20 deletions

diff --git a/src/mongo/util/queue.h b/src/mongo/util/queue.h
index 9f0870e9a82..a052cd1346e 100644
--- a/src/mongo/util/queue.h
+++ b/src/mongo/util/queue.h
@@ -36,15 +36,11 @@
 #include "mongo/base/disallow_copying.h"
 #include "mongo/stdx/chrono.h"
 #include "mongo/stdx/condition_variable.h"
+#include "mongo/stdx/functional.h"
 #include "mongo/stdx/mutex.h"
 
 namespace mongo {
 
-template <typename T>
-size_t _getSizeDefault(const T& t) {
-    return 1;
-}
-
 /**
  * Simple blocking queue with optional max size (by count or custom sizing function).
  * A custom sizing function can optionally be given.  By default the getSize function
@@ -56,13 +52,13 @@ size_t _getSizeDefault(const T& t) {
 template <typename T>
 class BlockingQueue {
     MONGO_DISALLOW_COPYING(BlockingQueue);
-    typedef size_t (*getSizeFunc)(const T& t);
 
 public:
-    BlockingQueue()
-        : _maxSize(std::numeric_limits<std::size_t>::max()), _getSize(&_getSizeDefault) {}
-    BlockingQueue(size_t size) : _maxSize(size), _getSize(&_getSizeDefault) {}
-    BlockingQueue(size_t size, getSizeFunc f) : _maxSize(size), _getSize(f) {}
+    using GetSizeFn = stdx::function<size_t(const T&)>;
+
+    BlockingQueue() : BlockingQueue(std::numeric_limits<std::size_t>::max()) {}
+    BlockingQueue(size_t size) : BlockingQueue(size, [](const T&) { return 1; }) {}
+    BlockingQueue(size_t size, GetSizeFn f) : _maxSize(size), _getSize(f) {}
 
     void pushEvenIfFull(T const& t) {
         stdx::unique_lock<stdx::mutex> lk(_lock);
@@ -82,21 +78,28 @@ public:
      *
      * NOTE: Should only be used in a single producer case.
      */
-    void pushAllNonBlocking(std::vector<T>& objs) {
-        if (objs.empty()) {
+    template <typename Container>
+    void pushAllNonBlocking(const Container& objs) {
+        pushAllNonBlocking(std::begin(objs), std::end(objs));
+    }
+
+    template <typename Iterator>
+    void pushAllNonBlocking(Iterator begin, Iterator end) {
+        if (begin == end) {
             return;
         }
 
         stdx::unique_lock<stdx::mutex> lk(_lock);
         const auto startedEmpty = _queue.empty();
         _clearing = false;
-        std::for_each(objs.begin(),
-                      objs.end(),
-                      [this](T& obj) {
-                          size_t tSize = _getSize(obj);
-                          _queue.push(obj);
-                          _currentSize += tSize;
-                      });
+
+        auto pushOne = [this](const T& obj) {
+            size_t tSize = _getSize(obj);
+            _queue.push(obj);
+            _currentSize += tSize;
+        };
+        std::for_each(begin, end, pushOne);
+
         if (startedEmpty) {
             _cvNoLongerEmpty.notify_one();
         }
@@ -269,7 +272,7 @@ private:
     std::queue<T> _queue;
     const size_t _maxSize;
     size_t _currentSize = 0;
-    getSizeFunc _getSize;
+    GetSizeFn _getSize;
     bool _clearing = false;
 
     stdx::condition_variable _cvNoLongerFull;