The review I asked for yesterday was for a header only library.
Stream that opens an HTTP GET and then acts like a normal C++ istream
This has some limitations as it downloads all the data as part of the constructor. This can lead to delays during start up if the file is large (usually during REST communications the size is small so the header only version is sufficient).
But if you want larger files and don't want a delay here is a version that downloads the data in the background asynchronously. If the stream buffer underflows the next block (which has been accumulated in the background) is quickly swapped into the buffer and the main thread can continue processing.
The disadvantage is that this is no longer a header only library and you will need to link against the library.
https://github.com/Loki-Astari/ThorsStream
ThorsStream.h
#ifndef THORSANVIL_STREAM_THOR_STREAM_H
#define THORSANVIL_STREAM_THOR_STREAM_H
// Note "ThorsSimpleStream.h" is up fro review in a seprate question
// https://codereview.stackexchange.com/questions/38402/http-stream-a-stream-that-opens-an-http-get-and-then-acts-like-a-normal-c-istr
#include "ThorsSimpleStream.h"
#include <thread>
#include <iostream>
namespace ThorsAnvil
{
namespace Stream
{
class IThorStream: public std::istream
{
// Class to handle the down-loading of all http connections in the background.
class ThorStreamManager
{
public:
// Everybody will use the same instance
static ThorStreamManager& defaultManager()
{
static ThorStreamManager defaultManager;
return defaultManager;
}
ThorStreamManager()
: finished(false)
, multi(curl_multi_init())
, streamThread(&ThorStreamManager::eventLoop, this, multi != NULL)
{
if (multi == NULL)
{ throw std::runtime_error("Failed to startup");
}
}
~ThorStreamManager()
{
// When destroying the object.
// Make sure we first take care of any threads.
{
std::unique_lock<std::mutex> lock(mutex);
finished = true;
}
cond.notify_one();
streamThread.join();
// Now we can clean up any resources.
curl_multi_cleanup(multi);
}
// Interface to add/remove http requests.
void addHTTPRequest(CURL* easy)
{
std::unique_lock<std::mutex> lock(mutex);
curl_multi_add_handle(multi, easy);
}
void delHTTPRequest(CURL* easy)
{
std::unique_lock<std::mutex> lock(mutex);
curl_multi_remove_handle(multi, easy);
}
private:
// Event loop is run by the thread.
// It handles all interactions with sockets.
void eventLoop(bool ok);
bool finished;
CURLM* multi;
std::mutex mutex;
std::condition_variable cond;
std::thread streamThread;
};
// The stream buffer.
class SocketStreamBuffer: public IThorSimpleStream::SimpleSocketStreamBuffer
{
public:
typedef IThorSimpleStream::SimpleSocketStreamBuffer::traits_type traits;
typedef traits::int_type int_type;
SocketStreamBuffer(std::string const& url, std::function<void()> markStreamBad)
: SimpleSocketStreamBuffer(url, false, false, markStreamBad)
{
/* Perform the request, res will get the return code */
ThorStreamManager::defaultManager().addHTTPRequest(curl);
}
~SocketStreamBuffer()
{
ThorStreamManager::defaultManager().delHTTPRequest(curl);
}
virtual int_type underflow()
{
{
std::unique_lock<std::mutex> lock(mutex);
empty = true;
}
curl_easy_pause(curl, CURLPAUSE_CONT);
{
std::unique_lock<std::mutex> lock(mutex);
cond.wait(lock, [&empty, &open](){return !(empty && open);});
}
return empty ? EOF : buffer[0];
}
// Used by ThorStreamManager to mark that the stream has
// finished downloading data.
virtual void markAsDone()
{
std::unique_lock<std::mutex> lock(mutex);
open = false;
cond.notify_one();
}
};
SocketStreamBuffer buffer;
public:
IThorStream(std::string const& url)
: std::istream(NULL)
, buffer(url, [this](){this->setstate(std::ios::badbit);})
{
std::istream::rdbuf(&buffer);
}
};
}
}
#endif
ThorsStream.cpp
#include "ThorsStream.h"
using namespace ThorsAnvil::Stream;
// Of OK is not true then the object
// will have thrown an exception so its
// its not safe to do anything but return
// immediately
void IThorStream::ThorStreamManager::eventLoop(bool ok)
{
// If multi failed to initialize in the constructor
// There is no point in continuing the thread.
if (!ok)
{ return;
}
// Can not let an exception escape a thread.
// If it did it would cause the application
// to terminate.
try
{
std::unique_lock<std::mutex> lock(mutex);
while(!finished)
{
// 1: Get the File descriptors that curl has ready.
int max_fd = 0;
fd_set read;
fd_set writ;
fd_set exec;
FD_ZERO(&read);
FD_ZERO(&writ);
FD_ZERO(&exec);
CURLMcode result = curl_multi_fdset(multi, &read, &writ, &exec, &max_fd);
if (result != CURLM_OK) {throw std::runtime_error("curl_multi_fdset: Failed");}
if (max_fd == 0)
{
// No file descriptors.
// There is no point doing anything.
// So wait until a new curl easy object is added.
// Then re-start the loop.
cond.wait(lock);
continue;
}
// 2: Calculate the timeout we can use in select/poll
timespec timeoutSpec = {0,0};
if (max_fd == -1)
{
// curl special case.
// We need to call curl_multi_perform() soon
// so use a 100 millisecond timeout
timeoutSpec.tv_sec = 0;
timeoutSpec.tv_nsec = 100 * 1000000; // convert milli to nano
}
else
{
long timeout;
result = curl_multi_timeout(multi, &timeout);
if (result != CURLM_OK) {throw std::runtime_error("curl_multi_timeout: Failed");}
if (timeout == -1)
{ timeout = 300;
}
timeoutSpec.tv_sec = timeout/1000;
timeoutSpec.tv_nsec = timeout%1000 * 1000000; // convert milli to nano
}
// 3: Wait for activity on any of the file descriptors.
int count;
do
{
lock.unlock();
count = pselect(max_fd+1, &read, &writ, &exec, &timeoutSpec, NULL);
lock.lock();
}
while ((count == -1) && (errno == EINTR));
if (count == -1) {throw std::runtime_error("pselect: Failed");}
// 4: Handle all the CURL calllback routines.
int running_handles;
do
{
result = curl_multi_perform(multi, &running_handles);
}
while(result == CURLM_CALL_MULTI_PERFORM);
if (result != CURLM_OK) {throw std::runtime_error("curl_multi_perform: Failed");}
// 5: Handle all the messages generated by curl.
// Currently this is simply the fact that a stream has closed.
CURLMsg* message;
int msgs_in_queue;
while((message = curl_multi_info_read(multi, &msgs_in_queue)) != NULL)
{
if (message->msg == CURLMSG_DONE)
{
CURL* easy = message->easy_handle;
char* priv;
if (curl_easy_getinfo(easy, CURLINFO_PRIVATE, &priv) == CURLE_OK && priv != NULL)
{
IThorStream::SocketStreamBuffer* owner = reinterpret_cast<IThorStream::SocketStreamBuffer*>(priv);
owner->markAsDone();
}
}
}
}
}
catch(std::exception const& e)
{
// Log error:
std::cerr << "Exception in Thread:\n\t" << e.what() << "\n";
}
catch(...)
{
// Log error:
std::cerr << "Exception in Thread:\n\tUNKNOW (...)\n";
}
// Ignore error
}
Unfortunately I used the select() version of the CURL interface. I have plans to re-write this to use the poll() version of the interface. But that's another day.
A simple example using the above stream and my previous JSON parser library to connect to a REST interface and download data directly into a C++ class object without parsing the JSON directly.
