convert DefaultFileSource to use util::Thread<>

30 files changed, 164 insertions, 243 deletions

diff --git a/android/cpp/native_map_view.cpp b/android/cpp/native_map_view.cpp
index 340bbae10b..b51e5cbc67 100644
--- a/android/cpp/native_map_view.cpp
+++ b/android/cpp/native_map_view.cpp
@@ -59,7 +59,7 @@ NativeMapView::NativeMapView(JNIEnv *env, jobject obj_)
     : mbgl::View(*this),
       fileCache(mbgl::android::cachePath + "/mbgl-cache.db"),
       fileSource(fileCache),
-      map(*this, fileSource) {
+      map(*this, *fileSource) {
     mbgl::Log::Debug(mbgl::Event::Android, "NativeMapView::NativeMapView");
 
     assert(env != nullptr);
@@ -145,7 +145,7 @@ void NativeMapView::notify() {
 
 mbgl::Map &NativeMapView::getMap() { return map; }
 
-mbgl::DefaultFileSource &NativeMapView::getFileSource() { return fileSource; }
+mbgl::DefaultFileSource &NativeMapView::getFileSource() { return *fileSource; }
 
 bool NativeMapView::inEmulator() {
     // Detect if we are in emulator
diff --git a/include/mbgl/android/native_map_view.hpp b/include/mbgl/android/native_map_view.hpp
index 4a8d9365bf..817b6bc8bb 100644
--- a/include/mbgl/android/native_map_view.hpp
+++ b/include/mbgl/android/native_map_view.hpp
@@ -63,7 +63,7 @@ private:
     ANativeWindow *window = nullptr;
 
     mbgl::util::Thread<mbgl::SQLiteCache> fileCache;
-    mbgl::DefaultFileSource fileSource;
+    mbgl::util::Thread<mbgl::DefaultFileSource> fileSource;
     mbgl::Map map;
 
     EGLDisplay display = EGL_NO_DISPLAY;
diff --git a/include/mbgl/storage/default/asset_request.hpp b/include/mbgl/storage/default/asset_request.hpp
index c582c025fb..48f402fefe 100644
--- a/include/mbgl/storage/default/asset_request.hpp
+++ b/include/mbgl/storage/default/asset_request.hpp
@@ -9,7 +9,7 @@ class AssetRequest : public SharedRequestBase {
 public:
     AssetRequest(DefaultFileSource *source, const Resource &resource);
 
-    void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr);
+    void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr);
     void cancel();
 
 private:
diff --git a/include/mbgl/storage/default/http_request.hpp b/include/mbgl/storage/default/http_request.hpp
index 914e622f13..1cd38a8281 100644
--- a/include/mbgl/storage/default/http_request.hpp
+++ b/include/mbgl/storage/default/http_request.hpp
@@ -9,7 +9,7 @@ class HTTPRequest : public SharedRequestBase {
 public:
     HTTPRequest(DefaultFileSource *source, const Resource &resource);
 
-    void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr);
+    void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr);
     void cancel();
 
     void retryImmediately();
diff --git a/include/mbgl/storage/default/shared_request_base.hpp b/include/mbgl/storage/default/shared_request_base.hpp
index 59e38efc2f..9c5f87b33d 100644
--- a/include/mbgl/storage/default/shared_request_base.hpp
+++ b/include/mbgl/storage/default/shared_request_base.hpp
@@ -29,15 +29,14 @@ public:
     SharedRequestBase(DefaultFileSource *source_, const Resource &resource_)
         : resource(resource_), source(source_) {}
 
-    virtual void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr) = 0;
+    virtual void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr) = 0;
     virtual void cancel() = 0;
 
-    void notify(std::unique_ptr<Response> response, FileCache::Hint hint) {
+    void notify(std::shared_ptr<const Response> response, FileCache::Hint hint) {
         MBGL_VERIFY_THREAD(tid);
 
         if (source) {
-            source->notify(this, observers, std::shared_ptr<const Response>(std::move(response)),
-                           hint);
+            source->notify(this, observers, response, hint);
         }
     }
 
diff --git a/include/mbgl/storage/default_file_source.hpp b/include/mbgl/storage/default_file_source.hpp
index 7aab54f731..f6028217ea 100644
--- a/include/mbgl/storage/default_file_source.hpp
+++ b/include/mbgl/storage/default_file_source.hpp
@@ -3,25 +3,25 @@
 
 #include <mbgl/storage/file_source.hpp>
 #include <mbgl/storage/file_cache.hpp>
+#include <mbgl/util/run_loop.hpp>
 
 #include <set>
 #include <unordered_map>
-#include <thread>
-
-namespace mapbox { namespace util { template<typename... Types> class variant; } }
 
 namespace mbgl {
 
-namespace util { template <typename T> class AsyncQueue; }
-
 class SharedRequestBase;
 
-class DefaultFileSource : public FileSource {
-public:
+class DefaultFileSource : public FileSource, protected util::RunLoop {
+    friend class util::Thread<DefaultFileSource>;
+    friend class SharedRequestBase;
+
+private:
     DefaultFileSource(FileCache *cache, const std::string &root = "");
-    DefaultFileSource(FileCache *cache, uv_loop_t *loop, const std::string &root = "");
     ~DefaultFileSource() override;
 
+public:
+    // FileSource API
     Request *request(const Resource &resource, uv_loop_t *loop, const Environment &env,
                      Callback callback) override;
     void cancel(Request *request) override;
@@ -29,37 +29,22 @@ public:
 
     void abort(const Environment &env) override;
 
+private:
     void notify(SharedRequestBase *sharedRequest, const std::set<Request *> &observers,
                 std::shared_ptr<const Response> response, FileCache::Hint hint);
+    SharedRequestBase *find(const Resource &resource);
+
+    void processAdd(Request* request);
+    void processCancel(Request* request);
+    void processResult(const Resource& resource, std::shared_ptr<const Response> response);
+    void processAbort(const Environment& env);
 
 public:
     const std::string assetRoot;
 
 private:
-    struct ActionDispatcher;
-    struct AddRequestAction;
-    struct RemoveRequestAction;
-    struct ResultAction;
-    struct StopAction;
-    struct AbortAction;
-    using Action = mapbox::util::variant<AddRequestAction, RemoveRequestAction, ResultAction,
-                                         StopAction, AbortAction>;
-    using Queue = util::AsyncQueue<Action>;
-
-    void process(AddRequestAction &action);
-    void process(RemoveRequestAction &action);
-    void process(ResultAction &action);
-    void process(StopAction &action);
-    void process(AbortAction &action);
-
-    SharedRequestBase *find(const Resource &resource);
-
     std::unordered_map<Resource, SharedRequestBase *, Resource::Hash> pending;
-
-    uv_loop_t *loop = nullptr;
     FileCache *cache = nullptr;
-    Queue *queue = nullptr;
-    std::thread thread;
 };
 
 }
diff --git a/include/mbgl/util/run_loop.hpp b/include/mbgl/util/run_loop.hpp
index 5aefcd1d3e..9bed417c38 100644
--- a/include/mbgl/util/run_loop.hpp
+++ b/include/mbgl/util/run_loop.hpp
@@ -1,6 +1,7 @@
 #ifndef MBGL_UTIL_RUN_LOOP
 #define MBGL_UTIL_RUN_LOOP
 
+#include <memory>
 #include <mutex>
 #include <functional>
 #include <queue>
@@ -31,13 +32,16 @@ protected:
     // Called by the Thread<> wrapper to terminate this loop.
     void stop();
 
+    // Obtain the underlying loop object in case you want to attach additional listeners.
+    uv::loop& loop() { return *runloop; };
+
 private:
     // Invokes function in the run loop.
     void process();
 
 public:
     // Schedules a function to be executed as part of this run loop.
-    void invoke(std::function<void()> fn);
+    void invoke(std::function<void()>&& fn);
 
 private:
     const std::unique_ptr<uv::loop> runloop;
diff --git a/linux/main.cpp b/linux/main.cpp
index 71c2424af5..d91aa7dec4 100644
--- a/linux/main.cpp
+++ b/linux/main.cpp
@@ -67,8 +67,8 @@ int main(int argc, char *argv[]) {
     view = mbgl::util::make_unique<GLFWView>();
 
     mbgl::util::Thread<mbgl::SQLiteCache> cache("/tmp/mbgl-cache.db");
-    mbgl::DefaultFileSource fileSource(cache);
-    mbgl::Map map(*view, fileSource);
+    mbgl::util::Thread<mbgl::DefaultFileSource> fileSource(cache);
+    mbgl::Map map(*view, *fileSource);
 
     // Load settings
     mbgl::Settings_JSON settings;
diff --git a/macosx/main.mm b/macosx/main.mm
index fb12282e28..d0f3282f8a 100644
--- a/macosx/main.mm
+++ b/macosx/main.mm
@@ -105,8 +105,8 @@ int main() {
     GLFWView view;
 
     mbgl::util::Thread<mbgl::SQLiteCache> cache(defaultCacheDatabase());
-    mbgl::DefaultFileSource fileSource(cache);
-    mbgl::Map map(view, fileSource);
+    mbgl::util::Thread<mbgl::DefaultFileSource> fileSource(cache);
+    mbgl::Map map(view, *fileSource);
 
     URLHandler *handler = [[URLHandler alloc] init];
     [handler setMap:&map];
diff --git a/platform/darwin/http_request_nsurl.mm b/platform/darwin/http_request_nsurl.mm
index 7be5f226fd..773f83e5d7 100644
--- a/platform/darwin/http_request_nsurl.mm
+++ b/platform/darwin/http_request_nsurl.mm
@@ -47,7 +47,7 @@ class HTTPNSURLContext;
 
 class HTTPRequestImpl {
 public:
-    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::unique_ptr<Response> response);
+    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::shared_ptr<const Response> response);
     ~HTTPRequestImpl();
 
     void cancel();
@@ -66,7 +66,7 @@ private:
     HTTPRequest *request = nullptr;
     NSURLSessionDataTask *task = nullptr;
     std::unique_ptr<Response> response;
-    std::unique_ptr<Response> existingResponse;
+    const std::shared_ptr<const Response> existingResponse;
     ResponseStatus status = ResponseStatus::PermanentError;
     uv_async_t *async = nullptr;
     int attempts = 0;
@@ -118,10 +118,10 @@ HTTPNSURLContext::~HTTPNSURLContext() {
 // -------------------------------------------------------------------------------------------------
 
 HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop,
-                                 std::unique_ptr<Response> existingResponse_)
+                                 std::shared_ptr<const Response> existingResponse_)
     : context(HTTPNSURLContext::Get(loop)),
       request(request_),
-      existingResponse(std::move(existingResponse_)),
+      existingResponse(existingResponse_),
       async(new uv_async_t) {
     assert(request);
     context->addRequest(request);
@@ -326,12 +326,12 @@ void HTTPRequestImpl::handleResult(NSData *data, NSURLResponse *res, NSError *er
 
         if (responseCode == 304) {
             if (existingResponse) {
-                // We're going to reuse the old response object, but need to copy over the new
-                // expires value (if possible).
-                std::swap(response, existingResponse);
-                if (existingResponse->expires) {
-                    response->expires = existingResponse->expires;
-                }
+                // We're going to copy over the existing response's data.
+                response->status = existingResponse->status;
+                response->message = existingResponse->message;
+                response->modified = existingResponse->modified;
+                response->etag = existingResponse->etag;
+                response->data = existingResponse->data;
                 status = ResponseStatus::NotModified;
             } else {
                 // This is an unsolicited 304 response and should only happen on malfunctioning
@@ -409,11 +409,11 @@ HTTPRequest::~HTTPRequest() {
     }
 }
 
-void HTTPRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void HTTPRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     assert(!ptr);
-    ptr = new HTTPRequestImpl(this, loop, std::move(response));
+    ptr = new HTTPRequestImpl(this, loop, response);
 }
 
 void HTTPRequest::retryImmediately() {
diff --git a/platform/default/asset_request_fs.cpp b/platform/default/asset_request_fs.cpp
index 5ebde2d888..10d5f33a31 100644
--- a/platform/default/asset_request_fs.cpp
+++ b/platform/default/asset_request_fs.cpp
@@ -220,7 +220,7 @@ AssetRequest::~AssetRequest() {
     }
 }
 
-void AssetRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void AssetRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     // We're ignoring the existing response if any.
diff --git a/platform/default/asset_request_zip.cpp b/platform/default/asset_request_zip.cpp
index fd27c4b959..85ff0aede4 100644
--- a/platform/default/asset_request_zip.cpp
+++ b/platform/default/asset_request_zip.cpp
@@ -288,7 +288,7 @@ AssetRequest::~AssetRequest() {
     }
 }
 
-void AssetRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void AssetRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     // We're ignoring the existing response if any.
diff --git a/platform/default/http_request_curl.cpp b/platform/default/http_request_curl.cpp
index a7ec162aa4..707465a3dc 100644
--- a/platform/default/http_request_curl.cpp
+++ b/platform/default/http_request_curl.cpp
@@ -95,7 +95,7 @@ class HTTPRequestImpl {
     MBGL_STORE_THREAD(tid)
 
 public:
-    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::unique_ptr<Response> response);
+    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::shared_ptr<const Response> response);
     ~HTTPRequestImpl();
 
     void handleResult(CURLcode code);
@@ -123,7 +123,7 @@ private:
     std::unique_ptr<Response> response;
 
     // In case of revalidation requests, this will store the old response.
-    std::unique_ptr<Response> existingResponse;
+    const std::shared_ptr<const Response> existingResponse;
 
     CURL *handle = nullptr;
     curl_slist *headers = nullptr;
@@ -425,10 +425,10 @@ static CURLcode sslctx_function(CURL * /* curl */, void *sslctx, void * /* parm
 }
 #endif
 
-HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop, std::unique_ptr<Response> response_)
+HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop, std::shared_ptr<const Response> response_)
     : context(HTTPCURLContext::Get(loop)),
       request(request_),
-      existingResponse(std::move(response_)),
+      existingResponse(response_),
       handle(context->getHandle()) {
     assert(request);
     context->addRequest(request);
@@ -688,12 +688,12 @@ void HTTPRequestImpl::handleResult(CURLcode code) {
 
         if (responseCode == 304) {
             if (existingResponse) {
-                // We're going to reuse the old response object, but need to copy over the new
-                // expires value (if possible).
-                std::swap(response, existingResponse);
-                if (existingResponse->expires) {
-                    response->expires = existingResponse->expires;
-                }
+                // We're going to copy over the existing response's data.
+                response->status = existingResponse->status;
+                response->message = existingResponse->message;
+                response->modified = existingResponse->modified;
+                response->etag = existingResponse->etag;
+                response->data = existingResponse->data;
                 return finish(ResponseStatus::NotModified);
             } else {
                 // This is an unsolicited 304 response and should only happen on malfunctioning
@@ -734,11 +734,11 @@ HTTPRequest::~HTTPRequest() {
     }
 }
 
-void HTTPRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void HTTPRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     assert(!ptr);
-    ptr = new HTTPRequestImpl(this, loop, std::move(response));
+    ptr = new HTTPRequestImpl(this, loop, response);
 }
 
 void HTTPRequest::retryImmediately() {
diff --git a/platform/ios/MGLMapView.mm b/platform/ios/MGLMapView.mm
index 662b371b8c..8281b6ec50 100644
--- a/platform/ios/MGLMapView.mm
+++ b/platform/ios/MGLMapView.mm
@@ -113,7 +113,7 @@ std::chrono::steady_clock::duration secondsAsDuration(float duration)
 mbgl::Map *mbglMap = nullptr;
 MBGLView *mbglView = nullptr;
 mbgl::util::Thread<mbgl::SQLiteCache> *mbglFileCache = nullptr;
-mbgl::DefaultFileSource *mbglFileSource = nullptr;
+mbgl::util::Thread<mbgl::DefaultFileSource> *mbglFileSource = nullptr;
 
 - (instancetype)initWithFrame:(CGRect)frame
 {
@@ -271,8 +271,8 @@ mbgl::DefaultFileSource *mbglFileSource = nullptr;
     //
     mbglView = new MBGLView(self);
     mbglFileCache  = new mbgl::util::Thread<mbgl::SQLiteCache>(defaultCacheDatabase());
-    mbglFileSource = new mbgl::DefaultFileSource(*mbglFileCache);
-    mbglMap = new mbgl::Map(*mbglView, *mbglFileSource);
+    mbglFileSource = new mbgl::util::Thread<mbgl::DefaultFileSource>(*mbglFileCache);
+    mbglMap = new mbgl::Map(*mbglView, **mbglFileSource);
     mbglView->resize(self.bounds.size.width, self.bounds.size.height, _glView.contentScaleFactor, _glView.drawableWidth, _glView.drawableHeight);
 
     // Notify map object when network reachability status changes.
diff --git a/src/mbgl/storage/default_file_source.cpp b/src/mbgl/storage/default_file_source.cpp
index 841a56bcef..0ee31b7fe3 100644
--- a/src/mbgl/storage/default_file_source.cpp
+++ b/src/mbgl/storage/default_file_source.cpp
@@ -6,10 +6,7 @@
 #include <mbgl/storage/response.hpp>
 #include <mbgl/platform/platform.hpp>
 
-#include <mbgl/util/async_queue.hpp>
-#include <mbgl/util/util.hpp>
-
-#include <mbgl/util/variant.hpp>
+#include <mbgl/util/uv_detail.hpp>
 #include <mbgl/util/chrono.hpp>
 #include <mbgl/platform/log.hpp>
 
@@ -30,72 +27,15 @@ namespace algo = boost::algorithm;
 
 namespace mbgl {
 
-struct DefaultFileSource::ActionDispatcher {
-    DefaultFileSource &fileSource;
-    template <typename T> void operator()(T &t) { fileSource.process(t); }
-};
-
-struct DefaultFileSource::AddRequestAction {
-    Request *const request;
-};
-
-struct DefaultFileSource::RemoveRequestAction {
-    Request *const request;
-};
-
-struct DefaultFileSource::ResultAction {
-    const Resource resource;
-    std::unique_ptr<Response> response;
-};
-
-struct DefaultFileSource::StopAction {
-};
-
-struct DefaultFileSource::AbortAction {
-    const Environment &env;
-};
-
-
-DefaultFileSource::DefaultFileSource(FileCache *cache_, const std::string &root)
-    : assetRoot(root.empty() ? platform::assetRoot() : root),
-      loop(uv_loop_new()),
-      cache(cache_),
-      queue(new Queue(loop, [this](Action &action) {
-          mapbox::util::apply_visitor(ActionDispatcher{*this}, action);
-      })),
-      thread([this]() {
+DefaultFileSource::DefaultFileSource(FileCache* cache_, const std::string& root)
+    : assetRoot(root.empty() ? platform::assetRoot() : root), cache(cache_) {
 #ifdef __APPLE__
-          pthread_setname_np("FileSource");
+    pthread_setname_np("FileSource");
 #endif
-          uv_run(loop, UV_RUN_DEFAULT);
-      }) {
-}
-
-DefaultFileSource::DefaultFileSource(FileCache *cache_, uv_loop_t *loop_, const std::string &root)
-    : assetRoot(root.empty() ? platform::assetRoot() : root),
-      loop(loop_),
-      cache(cache_),
-      queue(new Queue(loop, [this](Action &action) {
-          mapbox::util::apply_visitor(ActionDispatcher{*this}, action);
-      })) {
-    // Make sure that the queue doesn't block the loop from exiting.
-    queue->unref();
 }
 
 DefaultFileSource::~DefaultFileSource() {
     MBGL_VERIFY_THREAD(tid);
-
-    if (thread.joinable()) {
-        if (queue) {
-            queue->send(StopAction{ });
-        }
-        thread.join();
-        uv_loop_delete(loop);
-    } else {
-        // Assume that the loop we received is running in the current thread.
-        StopAction action {};
-        process(action);
-    }
 }
 
 SharedRequestBase *DefaultFileSource::find(const Resource &resource) {
@@ -110,18 +50,20 @@ SharedRequestBase *DefaultFileSource::find(const Resource &resource) {
     return nullptr;
 }
 
-Request *DefaultFileSource::request(const Resource &resource, uv_loop_t *l, const Environment &env,
+Request* DefaultFileSource::request(const Resource& resource,
+                                    uv_loop_t* l,
+                                    const Environment& env,
                                     Callback callback) {
     auto req = new Request(resource, l, env, std::move(callback));
 
     // This function can be called from any thread. Make sure we're executing the actual call in the
-    // file source loop by sending it over the queue. It will be processed in processAction().
-    queue->send(AddRequestAction{ req });
+    // file source loop by sending it over the queue.
+    invoke([this, req] { processAdd(req); });
+
     return req;
 }
 
-void DefaultFileSource::request(const Resource &resource, const Environment &env,
-                                Callback callback) {
+void DefaultFileSource::request(const Resource& resource, const Environment& env, Callback callback) {
     request(resource, nullptr, env, std::move(callback));
 }
 
@@ -129,17 +71,16 @@ void DefaultFileSource::cancel(Request *req) {
     req->cancel();
 
     // This function can be called from any thread. Make sure we're executing the actual call in the
-    // file source loop by sending it over the queue. It will be processed in processAction().
-    queue->send(RemoveRequestAction{ req });
+    // file source loop by sending it over the queue.
+    invoke([this, req] { processCancel(req); });
 }
 
 void DefaultFileSource::abort(const Environment &env) {
-    queue->send(AbortAction{ env });
+    invoke([this, &env] { processAbort(env); });
 }
 
-
-void DefaultFileSource::process(AddRequestAction &action) {
-    const Resource &resource = action.request->resource;
+void DefaultFileSource::processAdd(Request* request) {
+    const Resource &resource = request->resource;
 
     // We're adding a new Request.
     SharedRequestBase *sharedRequest = find(resource);
@@ -151,36 +92,32 @@ void DefaultFileSource::process(AddRequestAction &action) {
             sharedRequest = new HTTPRequest(this, resource);
         }
 
-        // Make sure the loop stays alive when we're not running the file source in it's own thread.
-        if (!thread.joinable() && pending.empty()) {
-            queue->ref();
-        }
-
         const bool inserted = pending.emplace(resource, sharedRequest).second;
         assert(inserted);
         (void (inserted)); // silence unused variable warning on Release builds.
 
         // But first, we're going to start querying the database if it exists.
         if (!cache) {
-            sharedRequest->start(loop);
+            sharedRequest->start(loop().get());
         } else {
             // Otherwise, first check the cache for existing data so that we can potentially
             // revalidate the information without having to redownload everything.
             cache->get(resource, [this, resource](std::unique_ptr<Response> response) {
-                queue->send(ResultAction { resource, std::move(response) });
+                std::shared_ptr<const Response> sharedResponse = std::move(response);
+                invoke([this, resource, sharedResponse] { processResult(resource, sharedResponse); });
             });
         }
     }
-    sharedRequest->subscribe(action.request);
+    sharedRequest->subscribe(request);
 }
 
-void DefaultFileSource::process(RemoveRequestAction &action) {
-    SharedRequestBase *sharedRequest = find(action.request->resource);
+void DefaultFileSource::processCancel(Request* request) {
+    SharedRequestBase *sharedRequest = find(request->resource);
     if (sharedRequest) {
         // If the number of dependent requests of the SharedRequestBase drops to zero, the
         // unsubscribe callback triggers the removal of the SharedRequestBase pointer from the list
         // of pending requests and initiates cancelation.
-        sharedRequest->unsubscribe(action.request);
+        sharedRequest->unsubscribe(request);
     } else {
         // There is no request for this URL anymore. Likely, the request already completed
         // before we got around to process the cancelation request.
@@ -188,28 +125,28 @@ void DefaultFileSource::process(RemoveRequestAction &action) {
 
     // Send a message back to the requesting thread and notify it that this request has been
     // canceled and is now safe to be deleted.
-    action.request->destruct();
+    request->destruct();
 }
 
-void DefaultFileSource::process(ResultAction &action) {
-    SharedRequestBase *sharedRequest = find(action.resource);
+void DefaultFileSource::processResult(const Resource& resource, std::shared_ptr<const Response> response) {
+    SharedRequestBase *sharedRequest = find(resource);
     if (sharedRequest) {
-        if (action.response) {
+        if (response) {
             // This entry was stored in the cache. Now determine if we need to revalidate.
             const int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
                                     SystemClock::now().time_since_epoch()).count();
-            if (action.response->expires > now) {
+            if (response->expires > now) {
                 // The response is fresh. We're good to notify the caller.
-                sharedRequest->notify(std::move(action.response), FileCache::Hint::No);
+                sharedRequest->notify(response, FileCache::Hint::No);
                 sharedRequest->cancel();
                 return;
             } else {
                 // The cached response is stale. Now run the real request.
-                sharedRequest->start(loop, std::move(action.response));
+                sharedRequest->start(loop().get(), response);
             }
         } else {
             // There is no response. Now run the real request.
-            sharedRequest->start(loop);
+            sharedRequest->start(loop().get());
         }
     } else {
         // There is no request for this URL anymore. Likely, the request was canceled
@@ -217,19 +154,8 @@ void DefaultFileSource::process(ResultAction &action) {
     }
 }
 
-// A stop action means the file source is about to be destructed. We need to cancel all requests
-// for all environments.
-void DefaultFileSource::process(StopAction &) {
-    // There may not be any pending requests in this file source anymore. You must terminate all
-    // Map objects before deleting the FileSource.
-    assert(pending.empty());
-    assert(queue);
-    queue->stop();
-    queue = nullptr;
-}
-
 // Aborts all requests that are part of the current environment.
-void DefaultFileSource::process(AbortAction &action) {
+void DefaultFileSource::processAbort(const Environment& env) {
     // Construct a cancellation response.
     auto res = util::make_unique<Response>();
     res->status = Response::Error;
@@ -239,7 +165,7 @@ void DefaultFileSource::process(AbortAction &action) {
     // Iterate through all pending requests and remove them in case they're abandoned.
     util::erase_if(pending, [&](const std::pair<Resource, SharedRequestBase *> &it) -> bool {
         // Obtain all pending requests that are in the current environment.
-        const auto aborted = it.second->removeAllInEnvironment(action.env);
+        const auto aborted = it.second->removeAllInEnvironment(env);
 
         // Notify all observers.
         for (auto req : aborted) {
@@ -274,11 +200,6 @@ void DefaultFileSource::notify(SharedRequestBase *sharedRequest,
             req->notify(response);
         }
     }
-
-    if (!thread.joinable() && pending.empty()) {
-        // When there are no pending requests, we're going to allow the queue to stop.
-        queue->unref();
-    }
 }
 
 }
diff --git a/src/mbgl/util/run_loop.cpp b/src/mbgl/util/run_loop.cpp
index 3a7cf597ff..dba75fcfcf 100644
--- a/src/mbgl/util/run_loop.cpp
+++ b/src/mbgl/util/run_loop.cpp
@@ -44,9 +44,9 @@ void RunLoop::process() {
     }
 }
 
-void RunLoop::invoke(std::function<void()> fn) {
+void RunLoop::invoke(std::function<void()>&& fn) {
     if (fn) {
-        critical_section(runloopMutex, [this, &fn] { runloopQueue.push(fn); });
+        critical_section(runloopMutex, [this, &fn] { runloopQueue.emplace(std::move(fn)); });
         runloopAsync->send();
     }
 }
diff --git a/test/headless/headless.cpp b/test/headless/headless.cpp
index c47b9349a8..76c1264333 100644
--- a/test/headless/headless.cpp
+++ b/test/headless/headless.cpp
@@ -4,6 +4,7 @@
 #include <mbgl/map/map.hpp>
 #include <mbgl/util/image.hpp>
 #include <mbgl/util/std.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <mbgl/util/io.hpp>
 #include <rapidjson/document.h>
@@ -139,8 +140,8 @@ TEST_P(HeadlessTest, render) {
         }
 
         HeadlessView view(display);
-        mbgl::DefaultFileSource fileSource(nullptr);
-        Map map(view, fileSource);
+        util::Thread<DefaultFileSource> fileSource(nullptr);
+        Map map(view, *fileSource);
 
         map.setClasses(classes);
         map.setStyleJSON(style, "test/suite");
diff --git a/test/storage/cache_response.cpp b/test/storage/cache_response.cpp
index f5d7c4bf3c..46083e2f93 100644
--- a/test/storage/cache_response.cpp
+++ b/test/storage/cache_response.cpp
@@ -12,12 +12,12 @@ TEST_F(Storage, CacheResponse) {
     using namespace mbgl;
 
     util::Thread<SQLiteCache> cache(":memory:");
-    DefaultFileSource fs(cache, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(cache);
 
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/cache" };
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request(resource, uv_default_loop(), env, [&](const Response &res) {
+    fs->request(resource, uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Response 1", res.data);
         EXPECT_LT(0, res.expires);
@@ -25,7 +25,7 @@ TEST_F(Storage, CacheResponse) {
         EXPECT_EQ("", res.etag);
         EXPECT_EQ("", res.message);
 
-        fs.request(resource, uv_default_loop(), env, [&, res](const Response &res2) {
+        fs->request(resource, uv_default_loop(), env, [&, res](const Response &res2) {
             EXPECT_EQ(res.status, res2.status);
             EXPECT_EQ(res.data, res2.data);
             EXPECT_EQ(res.expires, res2.expires);
diff --git a/test/storage/cache_revalidate.cpp b/test/storage/cache_revalidate.cpp
index 29f25eb775..901c151f14 100644
--- a/test/storage/cache_revalidate.cpp
+++ b/test/storage/cache_revalidate.cpp
@@ -14,12 +14,12 @@ TEST_F(Storage, CacheRevalidate) {
     using namespace mbgl;
 
     util::Thread<SQLiteCache> cache(":memory:");
-    DefaultFileSource fs(cache);
+    util::Thread<DefaultFileSource> fs(cache);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
     const Resource revalidateSame { Resource::Unknown, "http://127.0.0.1:3000/revalidate-same" };
-    fs.request(revalidateSame, uv_default_loop(), env, [&](const Response &res) {
+    fs->request(revalidateSame, uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Response", res.data);
         EXPECT_EQ(0, res.expires);
@@ -27,7 +27,7 @@ TEST_F(Storage, CacheRevalidate) {
         EXPECT_EQ("snowfall", res.etag);
         EXPECT_EQ("", res.message);
 
-        fs.request(revalidateSame, uv_default_loop(), env, [&, res](const Response &res2) {
+        fs->request(revalidateSame, uv_default_loop(), env, [&, res](const Response &res2) {
             EXPECT_EQ(Response::Successful, res2.status);
             EXPECT_EQ("Response", res2.data);
             // We use this to indicate that a 304 reply came back.
@@ -43,7 +43,7 @@ TEST_F(Storage, CacheRevalidate) {
 
     const Resource revalidateModified{ Resource::Unknown,
                                        "http://127.0.0.1:3000/revalidate-modified" };
-    fs.request(revalidateModified, uv_default_loop(), env, [&](const Response &res) {
+    fs->request(revalidateModified, uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Response", res.data);
         EXPECT_EQ(0, res.expires);
@@ -51,7 +51,7 @@ TEST_F(Storage, CacheRevalidate) {
         EXPECT_EQ("", res.etag);
         EXPECT_EQ("", res.message);
 
-        fs.request(revalidateModified, uv_default_loop(), env, [&, res](const Response &res2) {
+        fs->request(revalidateModified, uv_default_loop(), env, [&, res](const Response &res2) {
             EXPECT_EQ(Response::Successful, res2.status);
             EXPECT_EQ("Response", res2.data);
             // We use this to indicate that a 304 reply came back.
@@ -65,7 +65,7 @@ TEST_F(Storage, CacheRevalidate) {
     });
 
     const Resource revalidateEtag { Resource::Unknown, "http://127.0.0.1:3000/revalidate-etag" };
-    fs.request(revalidateEtag, uv_default_loop(), env, [&](const Response &res) {
+    fs->request(revalidateEtag, uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Response 1", res.data);
         EXPECT_EQ(0, res.expires);
@@ -73,7 +73,7 @@ TEST_F(Storage, CacheRevalidate) {
         EXPECT_EQ("response-1", res.etag);
         EXPECT_EQ("", res.message);
 
-        fs.request(revalidateEtag, uv_default_loop(), env, [&, res](const Response &res2) {
+        fs->request(revalidateEtag, uv_default_loop(), env, [&, res](const Response &res2) {
             EXPECT_EQ(Response::Successful, res2.status);
             EXPECT_EQ("Response 2", res2.data);
             EXPECT_EQ(0, res2.expires);
diff --git a/test/storage/directory_reading.cpp b/test/storage/directory_reading.cpp
index a955648462..4264129b22 100644
--- a/test/storage/directory_reading.cpp
+++ b/test/storage/directory_reading.cpp
@@ -3,6 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, AssetReadDirectory) {
     SCOPED_TEST(ReadDirectory)
@@ -10,14 +11,14 @@ TEST_F(Storage, AssetReadDirectory) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    util::Thread<DefaultFileSource> fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage" }, uv_default_loop(),
+    fs->request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage" }, uv_default_loop(),
                env, [&](const Response &res) {
         EXPECT_EQ(Response::Error, res.status);
         EXPECT_EQ(0ul, res.data.size());
diff --git a/test/storage/file_reading.cpp b/test/storage/file_reading.cpp
index cca072b27a..de834798de 100644
--- a/test/storage/file_reading.cpp
+++ b/test/storage/file_reading.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/platform/platform.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, AssetEmptyFile) {
     SCOPED_TEST(EmptyFile)
@@ -11,14 +12,14 @@ TEST_F(Storage, AssetEmptyFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    util::Thread<DefaultFileSource> fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/empty" }, uv_default_loop(),
+    fs->request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/empty" }, uv_default_loop(),
                env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ(0ul, res.data.size());
@@ -38,14 +39,14 @@ TEST_F(Storage, AssetNonEmptyFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    util::Thread<DefaultFileSource> fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/nonempty" },
+    fs->request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/nonempty" },
                uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ(16ul, res.data.size());
@@ -66,14 +67,14 @@ TEST_F(Storage, AssetNonExistentFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    util::Thread<DefaultFileSource> fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/does_not_exist" },
+    fs->request({ Resource::Unknown, "asset://TEST_DATA/fixtures/storage/does_not_exist" },
                uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Error, res.status);
         EXPECT_EQ(0ul, res.data.size());
diff --git a/test/storage/http_cancel.cpp b/test/storage/http_cancel.cpp
index d0dac8ccdb..33edbdf7c4 100644
--- a/test/storage/http_cancel.cpp
+++ b/test/storage/http_cancel.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/network_status.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <cmath>
 
@@ -12,15 +13,15 @@ TEST_F(Storage, HTTPCancel) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
     auto req =
-        fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
+        fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
                    [&](const Response &) { ADD_FAILURE() << "Callback should not be called"; });
 
-    fs.cancel(req);
+    fs->cancel(req);
     HTTPCancel.finish();
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);
@@ -31,15 +32,15 @@ TEST_F(Storage, HTTPCancelMultiple) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/test" };
 
-    auto req2 = fs.request(resource, uv_default_loop(), env, [&](const Response &) {
+    auto req2 = fs->request(resource, uv_default_loop(), env, [&](const Response &) {
         ADD_FAILURE() << "Callback should not be called";
     });
-    fs.request(resource, uv_default_loop(), env, [&](const Response &res) {
+    fs->request(resource, uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Hello World!", res.data);
         EXPECT_EQ(0, res.expires);
@@ -48,7 +49,7 @@ TEST_F(Storage, HTTPCancelMultiple) {
         EXPECT_EQ("", res.message);
         HTTPCancelMultiple.finish();
     });
-    fs.cancel(req2);
+    fs->cancel(req2);
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);
 }
diff --git a/test/storage/http_coalescing.cpp b/test/storage/http_coalescing.cpp
index 9eaea70e11..5164117f52 100644
--- a/test/storage/http_coalescing.cpp
+++ b/test/storage/http_coalescing.cpp
@@ -3,6 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, HTTPCoalescing) {
     SCOPED_TEST(HTTPCoalescing)
@@ -12,7 +13,7 @@ TEST_F(Storage, HTTPCoalescing) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -43,7 +44,7 @@ TEST_F(Storage, HTTPCoalescing) {
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/test" };
 
     for (int i = 0; i < total; i++) {
-        fs.request(resource, uv_default_loop(), env, complete);
+        fs->request(resource, uv_default_loop(), env, complete);
     }
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);
diff --git a/test/storage/http_environment.cpp b/test/storage/http_environment.cpp
index efd7b78ad4..b08a8071c4 100644
--- a/test/storage/http_environment.cpp
+++ b/test/storage/http_environment.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/network_status.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <cmath>
 
@@ -13,7 +14,7 @@ TEST_F(Storage, HTTPCancelEnvironment) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     // Create two fake environment pointers. The FileSource implementation treats these as opaque
     // pointers and doesn't reach into them.
@@ -23,7 +24,7 @@ TEST_F(Storage, HTTPCancelEnvironment) {
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/delayed" };
 
     // Environment 1
-    fs.request(resource, uv_default_loop(), env1, [&](const Response &res) {
+    fs->request(resource, uv_default_loop(), env1, [&](const Response &res) {
         // This environment gets aborted below. This means the request is marked as failing and
         // will return an error here.
         EXPECT_EQ(Response::Error, res.status);
@@ -36,7 +37,7 @@ TEST_F(Storage, HTTPCancelEnvironment) {
     });
 
     // Environment 2
-    fs.request(resource, uv_default_loop(), env2, [&](const Response &res) {
+    fs->request(resource, uv_default_loop(), env2, [&](const Response &res) {
         // The same request as above, but in a different environment which doesn't get aborted. This
         // means the request should succeed.
         EXPECT_EQ(Response::Successful, res.status);
@@ -48,7 +49,7 @@ TEST_F(Storage, HTTPCancelEnvironment) {
         HTTPRetainedEnvironment.finish();
     });
 
-    fs.abort(env1);
+    fs->abort(env1);
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);
 }
diff --git a/test/storage/http_error.cpp b/test/storage/http_error.cpp
index abaeff5396..bd5ef9811d 100644
--- a/test/storage/http_error.cpp
+++ b/test/storage/http_error.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/network_status.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <cmath>
 
@@ -20,13 +21,13 @@ TEST_F(Storage, HTTPError) {
     }, 500, 500);
     uv_unref((uv_handle_t *)&statusChange);
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
     auto start = uv_hrtime();
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/temporary-error" }, uv_default_loop(),
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/temporary-error" }, uv_default_loop(),
                env, [&](const Response &res) {
         const auto duration = double(uv_hrtime() - start) / 1e9;
         EXPECT_LT(1, duration) << "Backoff timer didn't wait 1 second";
@@ -41,7 +42,7 @@ TEST_F(Storage, HTTPError) {
         HTTPTemporaryError.finish();
     });
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3001/" }, uv_default_loop(), env,
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3001/" }, uv_default_loop(), env,
                [&](const Response &res) {
         const auto duration = double(uv_hrtime() - start) / 1e9;
         // 1.5 seconds == 4 retries, with a 500ms timeout (see above).
diff --git a/test/storage/http_header_parsing.cpp b/test/storage/http_header_parsing.cpp
index e4d86fcc27..df1fb1fc68 100644
--- a/test/storage/http_header_parsing.cpp
+++ b/test/storage/http_header_parsing.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/util/chrono.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <cmath>
 
@@ -13,11 +14,11 @@ TEST_F(Storage, HTTPHeaderParsing) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown,
+    fs->request({ Resource::Unknown,
                  "http://127.0.0.1:3000/test?modified=1420794326&expires=1420797926&etag=foo" },
                uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
@@ -32,7 +33,7 @@ TEST_F(Storage, HTTPHeaderParsing) {
     int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
                        SystemClock::now().time_since_epoch()).count();
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test?cachecontrol=max-age=120" },
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test?cachecontrol=max-age=120" },
                uv_default_loop(), env, [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Hello World!", res.data);
diff --git a/test/storage/http_load.cpp b/test/storage/http_load.cpp
index 2680daf93b..b037867233 100644
--- a/test/storage/http_load.cpp
+++ b/test/storage/http_load.cpp
@@ -3,13 +3,14 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, HTTPLoad) {
     SCOPED_TEST(HTTPLoad)
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -19,7 +20,7 @@ TEST_F(Storage, HTTPLoad) {
 
     std::function<void()> req = [&]() {
         const auto current = number++;
-        fs.request({ Resource::Unknown,
+        fs->request({ Resource::Unknown,
                      std::string("http://127.0.0.1:3000/load/") + std::to_string(current) },
                    uv_default_loop(), env, [&, current](const Response &res) {
             EXPECT_EQ(Response::Successful, res.status);
diff --git a/test/storage/http_noloop.cpp b/test/storage/http_noloop.cpp
index cd4ebeb2c4..0f75c106fe 100644
--- a/test/storage/http_noloop.cpp
+++ b/test/storage/http_noloop.cpp
@@ -4,13 +4,14 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/util/uv.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, HTTPNoLoop) {
     SCOPED_TEST(HTTPNoLoop)
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr);
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -22,7 +23,7 @@ TEST_F(Storage, HTTPNoLoop) {
         uv::close(as);
     });
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/temporary-error" }, nullptr, env,
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/temporary-error" }, nullptr, env,
                [&](const Response &res) {
         EXPECT_NE(uv_thread_self(), mainThread) << "Response was called in the same thread";
         EXPECT_EQ(Response::Successful, res.status);
diff --git a/test/storage/http_other_loop.cpp b/test/storage/http_other_loop.cpp
index 9ad673cf79..11fa2f324a 100644
--- a/test/storage/http_other_loop.cpp
+++ b/test/storage/http_other_loop.cpp
@@ -3,6 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, HTTPOtherLoop) {
     SCOPED_TEST(HTTPOtherLoop)
@@ -10,11 +11,11 @@ TEST_F(Storage, HTTPOtherLoop) {
     using namespace mbgl;
 
     // This file source launches a separate thread to do the processing.
-    DefaultFileSource fs(nullptr);
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
                [&](const Response &res) {
         EXPECT_EQ(Response::Successful, res.status);
         EXPECT_EQ("Hello World!", res.data);
diff --git a/test/storage/http_reading.cpp b/test/storage/http_reading.cpp
index a6a5775825..1ba34801cd 100644
--- a/test/storage/http_reading.cpp
+++ b/test/storage/http_reading.cpp
@@ -3,6 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, HTTPReading) {
     SCOPED_TEST(HTTPTest)
@@ -10,13 +11,13 @@ TEST_F(Storage, HTTPReading) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
     const auto mainThread = uv_thread_self();
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
                [&](const Response &res) {
         EXPECT_EQ(uv_thread_self(), mainThread);
         EXPECT_EQ(Response::Successful, res.status);
@@ -28,7 +29,7 @@ TEST_F(Storage, HTTPReading) {
         HTTPTest.finish();
     });
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/doesnotexist" }, uv_default_loop(),
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/doesnotexist" }, uv_default_loop(),
                env, [&](const Response &res) {
         EXPECT_EQ(uv_thread_self(), mainThread);
         EXPECT_EQ(Response::Error, res.status);
@@ -47,11 +48,11 @@ TEST_F(Storage, HTTPNoCallback) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, uv_default_loop(), env,
                nullptr);
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);
@@ -64,11 +65,11 @@ TEST_F(Storage, HTTPNoCallbackNoLoop) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    util::Thread<DefaultFileSource> fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, env, nullptr);
+    fs->request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, env, nullptr);
 
     uv_run(uv_default_loop(), UV_RUN_DEFAULT);