Support for streaming images in OpenGL format (bottom-up row order)

apps/SimpleStreamer/main.cpp

@@ -254,9 +254,6 @@ struct Image
         image.data.resize(image.width * image.height * 4);
         glReadPixels(0, 0, image.width, image.height, GL_RGBA, GL_UNSIGNED_BYTE,
                      (GLvoid*)image.data.data());
-
-        deflect::ImageWrapper::swapYAxis(image.data.data(), image.width,
-                                         image.height, 4);
         return image;
     }
 
@@ -292,6 +289,7 @@ bool send(const Image& image, const deflect::View view)
                                          : deflect::COMPRESSION_OFF;
     deflectImage.compressionQuality = deflectCompressionQuality;
     deflectImage.view = view;
+    deflectImage.rowOrder = deflect::RowOrder::bottom_up;
     return deflectStream->send(deflectImage).get();
 }
 

deflect/CMakeLists.txt

@@ -29,6 +29,7 @@ set(DEFLECT_HEADERS
   Socket.h
   SourceBuffer.h
   StreamPrivate.h
+  TaskBuilder.h
 )
 
 set(DEFLECT_SOURCES
@@ -48,6 +49,7 @@ set(DEFLECT_SOURCES
   Stream.cpp
   StreamPrivate.cpp
   StreamSendWorker.cpp
+  TaskBuilder.cpp
 )
 
 set(DEFLECT_LINK_LIBRARIES PRIVATE Qt5::Concurrent Qt5::Core Qt5::Network)

deflect/Frame.cpp

@@ -45,14 +45,30 @@ QSize Frame::computeDimensions() const
 {
     QSize size(0, 0);
 
-    for (size_t i = 0; i < segments.size(); ++i)
+    for (const auto& segment : segments)
     {
-        const deflect::SegmentParameters& params = segments[i].parameters;
+        const auto& params = segment.parameters;
         size.setWidth(std::max(size.width(), (int)(params.width + params.x)));
         size.setHeight(
             std::max(size.height(), (int)(params.height + params.y)));
     }
 
     return size;
 }
+
+RowOrder Frame::determineRowOrder() const
+{
+    if (segments.empty())
+        throw std::runtime_error("frame has no segements");
+
+    const auto frameRowOrder = segments[0].rowOrder;
+
+    for (const auto& segment : segments)
+    {
+        if (segment.rowOrder != frameRowOrder)
+            throw std::runtime_error("frame has incoherent row orders");
+    }
+
+    return frameRowOrder;
+}
 }

deflect/Frame.h

@@ -63,6 +63,12 @@ class Frame
 
     /** Get the total dimensions of this frame. */
     DEFLECT_API QSize computeDimensions() const;
+
+    /**
+     * @return the row order of all frame segments
+     * @throws std::runtime_error if not all segments have the same RowOrder
+     */
+    DEFLECT_API RowOrder determineRowOrder() const;
 };
 }
 

deflect/FrameDispatcher.cpp

@@ -63,12 +63,22 @@ class FrameDispatcher::Impl
 
         assert(!frame->segments.empty());
 
+        if (frame->determineRowOrder() == RowOrder::bottom_up)
+            mirrorSegmentsPositionsVertically(*frame);
+
         // receiver will request a new frame once this frame was consumed
         buffer.setAllowedToSend(false);
 
         return frame;
     }
 
+    void mirrorSegmentsPositionsVertically(Frame& frame) const
+    {
+        const auto height = frame.computeDimensions().height();
+        for (auto& s : frame.segments)
+            s.parameters.y = height - s.parameters.y - s.parameters.height;
+    }
+
     typedef std::map<QString, ReceiveBuffer> StreamBuffers;
     StreamBuffers streamBuffers;
     std::map<QString, size_t> observers;
@@ -143,17 +153,13 @@ void FrameDispatcher::processFrameFinished(const QString uri,
     try
     {
         buffer.finishFrameForSource(sourceIndex);
+        if (buffer.isAllowedToSend() && buffer.hasCompleteFrame())
+            emit sendFrame(_impl->consumeLatestFrame(uri));
     }
     catch (const std::runtime_error& e)
     {
-        std::cerr << "processFrameFinished got exception, closing stream: "
-                  << e.what() << std::endl;
-        emit bufferSizeExceeded(uri);
-        return;
+        emit pixelStreamException(uri, e.what());
     }
-
-    if (buffer.isAllowedToSend() && buffer.hasCompleteFrame())
-        emit sendFrame(_impl->consumeLatestFrame(uri));
 }
 
 void FrameDispatcher::requestFrame(const QString uri)
@@ -163,8 +169,15 @@ void FrameDispatcher::requestFrame(const QString uri)
 
     ReceiveBuffer& buffer = _impl->streamBuffers[uri];
     buffer.setAllowedToSend(true);
-    if (buffer.hasCompleteFrame())
-        emit sendFrame(_impl->consumeLatestFrame(uri));
+    try
+    {
+        if (buffer.hasCompleteFrame())
+            emit sendFrame(_impl->consumeLatestFrame(uri));
+    }
+    catch (const std::runtime_error& e)
+    {
+        emit pixelStreamException(uri, e.what());
+    }
 }
 
 void FrameDispatcher::deleteStream(const QString uri)

deflect/FrameDispatcher.h

@@ -58,7 +58,7 @@ class FrameDispatcher : public QObject
 
 public:
     /** Construct a dispatcher */
-    FrameDispatcher(QObject* parent);
+    FrameDispatcher(QObject* parent = nullptr);
 
     /** Destructor. */
     ~FrameDispatcher();
@@ -142,6 +142,14 @@ public slots:
      */
     void pixelStreamOpened(QString uri);
 
+    /**
+     * Notify that an exception occured and the stream should be closed.
+     *
+     * @param uri Identifier for the stream
+     * @param what The description of the exception that occured
+     */
+    void pixelStreamException(QString uri, QString what);
+
     /**
      * Notify that a pixel stream has been closed.
      *
@@ -156,13 +164,6 @@ public slots:
      */
     void sendFrame(deflect::FramePtr frame);
 
-    /**
-     * Notify that a pixel stream has exceeded its maximum allowed size.
-     *
-     * @param uri Identifier for the stream
-     */
-    void bufferSizeExceeded(QString uri);
-
 private:
     class Impl;
     std::unique_ptr<Impl> _impl;

deflect/ImageSegmenter.cpp

@@ -62,7 +62,7 @@ bool _isOnRightSideOfSideBySideImage(const Segment& segment)
 }
 }
 
-bool ImageSegmenter::generate(const ImageWrapper& image, const Handler& handler)
+bool ImageSegmenter::generate(const ImageWrapper& image, Handler handler)
 {
     if (image.compressionPolicy == COMPRESSION_ON)
         return _generateJpeg(image, handler);
@@ -240,6 +240,7 @@ Segments ImageSegmenter::_generateSegments(const ImageWrapper& image) const
         segment.view =
             image.view == View::side_by_side ? View::left_eye : image.view;
         segment.sourceImage = ℑ
+        segment.rowOrder = image.rowOrder;
         segments.push_back(segment);
     }
 

deflect/ImageSegmenter.h

@@ -75,8 +75,7 @@ class ImageSegmenter
      * @return true if all image handlers returned true, false on failure
      * @see setNominalSegmentDimensions()
      */
-    DEFLECT_API bool generate(const ImageWrapper& image,
-                              const Handler& handler);
+    DEFLECT_API bool generate(const ImageWrapper& image, Handler handler);
 
     /**
      * Set the nominal segment dimensions.

deflect/ImageWrapper.h

@@ -137,6 +137,22 @@ struct ImageWrapper
      */
     View view = View::mono;
 
+    /**
+     * The order of the image's data rows in memory.
+     *
+     * Set this value to bottom_up if the image data is stored following the
+     * OpenGL convention.
+     *
+     * All images that form a frame (possibly from multiple Streams) must have
+     * the same row order, otherwise the frame is invalid. This is because the
+     * frame's segments need to be reordered in addtion to flipping them
+     * individually.
+     *
+     * This is an alternative to calling swapYAxis() before sending.
+     * @version 1.7
+     */
+    RowOrder rowOrder = RowOrder::top_down;
+
     /**
      * Get the number of bytes per pixel based on the pixelFormat.
      * @version 1.0

deflect/MessageHeader.h

@@ -68,7 +68,8 @@ enum MessageType
     MESSAGE_TYPE_SIZE_HINTS = 13,
     MESSAGE_TYPE_DATA = 14,
     MESSAGE_TYPE_IMAGE_VIEW = 15,
-    MESSAGE_TYPE_OBSERVER_OPEN = 16
+    MESSAGE_TYPE_OBSERVER_OPEN = 16,
+    MESSAGE_TYPE_IMAGE_ROW_ORDER = 17
 };
 
 #define MESSAGE_HEADER_URI_LENGTH 64

deflect/Observer.cpp

@@ -100,7 +100,7 @@ bool Observer::registerForEvents(const bool exclusive)
         return true;
 
     // Send the bind message
-    if (!_impl->sendWorker.enqueueBindRequest(exclusive).get())
+    if (!_impl->bindEvents(exclusive).get())
     {
         std::cerr << "deflect::Stream::registerForEvents: sending bind message "
                   << "failed" << std::endl;
@@ -175,13 +175,11 @@ void Observer::setDisconnectedCallback(const std::function<void()> callback)
 
 void Observer::sendSizeHints(const SizeHints& hints)
 {
-    _impl->sendWorker.enqueueSizeHints(hints);
+    _impl->send(hints);
 }
 
 bool Observer::sendData(const char* data, const size_t count)
 {
-    return _impl->sendWorker
-        .enqueueData(QByteArray::fromRawData(data, int(count)))
-        .get();
+    return _impl->send(QByteArray::fromRawData(data, int(count))).get();
 }
 }

deflect/Segment.h

@@ -1,5 +1,5 @@
 /*********************************************************************/
-/* Copyright (c) 2013-2016, EPFL/Blue Brain Project                  */
+/* Copyright (c) 2013-2017, EPFL/Blue Brain Project                  */
 /*                          Raphael Dumusc <raphael.dumusc@epfl.ch>  */
 /* All rights reserved.                                              */
 /*                                                                   */
@@ -61,6 +61,8 @@ struct Segment
     /** Image data of the segment. */
     QByteArray imageData;
 
+    RowOrder rowOrder = RowOrder::top_down; //!< imageData row order
+
     /** @internal raw, uncompressed source image, used for compression */
     const ImageWrapper* sourceImage = nullptr;
 

deflect/Server.cpp

@@ -83,8 +83,11 @@ Server::Server(const int port)
             &Server::pixelStreamClosed);
     connect(_impl->frameDispatcher, &FrameDispatcher::sendFrame, this,
             &Server::receivedFrame);
-    connect(_impl->frameDispatcher, &FrameDispatcher::bufferSizeExceeded, this,
-            &Server::closePixelStream);
+    connect(_impl->frameDispatcher, &FrameDispatcher::pixelStreamException,
+            [this](const QString uri, const QString what) {
+                emit pixelStreamException(uri, what);
+                closePixelStream(uri);
+            });
 }
 
 Server::~Server()

deflect/Server.h

@@ -105,6 +105,16 @@ public slots:
      */
     void pixelStreamOpened(QString uri);
 
+    /**
+     * Notify that a stream has encountered an exception and will be closed.
+     *
+     * Used for error reporting.
+     *
+     * @param uri Identifier for the stream
+     * @param what The error message
+     */
+    void pixelStreamException(QString uri, QString what);
+
     /**
      * Notify that a pixel stream has been closed.
      *

deflect/ServerWorker.cpp

@@ -61,6 +61,7 @@ ServerWorker::ServerWorker(const int socketDescriptor)
     , _clientProtocolVersion{NETWORK_PROTOCOL_VERSION}
     , _registeredToEvents{false}
     , _activeView{View::mono}
+    , _activeRowOrder{RowOrder::top_down}
 {
     if (!_tcpSocket->setSocketDescriptor(socketDescriptor))
     {
@@ -266,6 +267,14 @@ void ServerWorker::_handleMessage(const MessageHeader& messageHeader,
         break;
     }
 
+    case MESSAGE_TYPE_IMAGE_ROW_ORDER:
+    {
+        const auto order = reinterpret_cast<const RowOrder*>(byteArray.data());
+        if (*order >= RowOrder::top_down && *order <= RowOrder::bottom_up)
+            _activeRowOrder = *order;
+        break;
+    }
+
     case MESSAGE_TYPE_BIND_EVENTS:
     case MESSAGE_TYPE_BIND_EVENTS_EX:
         if (_registeredToEvents)
@@ -311,6 +320,8 @@ void ServerWorker::_handlePixelStreamMessage(const QByteArray& message)
     segment.imageData =
         message.right(message.size() - sizeof(SegmentParameters));
     segment.view = _activeView;
+    segment.rowOrder = _activeRowOrder;
+
     emit(receivedSegment(_streamId, _sourceId, segment));
 }
 

deflect/ServerWorker.h

@@ -106,6 +106,7 @@ private slots:
     QQueue<Event> _events;
 
     View _activeView;
+    RowOrder _activeRowOrder;
 
     void _receiveMessage();
     MessageHeader _receiveMessageHeader();

deflect/Stream.cpp

@@ -61,16 +61,16 @@ Stream::~Stream()
 
 Stream::Future Stream::send(const ImageWrapper& image)
 {
-    return _impl->sendWorker.enqueueImage(image, false);
+    return _impl->sendImage(image, false);
 }
 
 Stream::Future Stream::finishFrame()
 {
-    return _impl->sendWorker.enqueueFinish();
+    return _impl->sendFinishFrame();
 }
 
 Stream::Future Stream::sendAndFinish(const ImageWrapper& image)
 {
-    return _impl->sendWorker.enqueueImage(image, true);
+    return _impl->sendImage(image, true);
 }
 }