summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/InputReader.h8
-rw-r--r--src/InputZeroMQReader.cpp9
2 files changed, 10 insertions, 7 deletions
diff --git a/src/InputReader.h b/src/InputReader.h
index b8c4fae..d2b5d8c 100644
--- a/src/InputReader.h
+++ b/src/InputReader.h
@@ -150,8 +150,6 @@ struct InputZeroMQThreadData
ThreadsafeQueue<std::shared_ptr<std::vector<uint8_t> > > *in_messages;
std::string uri;
unsigned max_queued_frames;
-
- bool running;
};
class InputZeroMQWorker
@@ -164,10 +162,13 @@ class InputZeroMQWorker
void Start(struct InputZeroMQThreadData* workerdata);
void Stop();
+
+ bool is_running(void) { return running; }
private:
+ bool running;
+
void RecvProcess(struct InputZeroMQThreadData* workerdata);
- bool running;
zmq::context_t zmqcontext; // is thread-safe
boost::thread recv_thread;
@@ -186,7 +187,6 @@ class InputZeroMQReader : public InputReader
InputZeroMQReader()
{
workerdata_.in_messages = &in_messages_;
- workerdata_.running = false;
}
~InputZeroMQReader()
diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp
index 70b0afc..58d76c9 100644
--- a/src/InputZeroMQReader.cpp
+++ b/src/InputZeroMQReader.cpp
@@ -84,6 +84,10 @@ int InputZeroMQReader::GetNextFrame(void* buffer)
{
const size_t framesize = 6144;
+ if (not worker_.is_running()) {
+ return 0;
+ }
+
std::shared_ptr<std::vector<uint8_t> > incoming;
/* Do some prebuffering because reads will happen in bursts
@@ -98,7 +102,7 @@ int InputZeroMQReader::GetNextFrame(void* buffer)
in_messages_.wait_and_pop(incoming);
}
- if (! workerdata_.running) {
+ if (not worker_.is_running()) {
throw zmq_input_overflow();
}
@@ -217,14 +221,13 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
subscriber.close();
- workerdata->running = false;
+ running = false;
workerdata->in_messages->notify();
}
void InputZeroMQWorker::Start(struct InputZeroMQThreadData* workerdata)
{
running = true;
- workerdata->running = true;
recv_thread = boost::thread(&InputZeroMQWorker::RecvProcess, this, workerdata);
}