AsyncStreams: Stop socket and file requests when close or finish events happens

Author: Pagghiu

Libraries/AsyncStreams/AsyncRequestStreams.cpp

@@ -40,29 +40,24 @@ template <typename AsyncReadRequest>
 SC::AsyncRequestReadableStream<AsyncReadRequest>::AsyncRequestReadableStream()
 {
     AsyncReadableStream::asyncRead.bind<AsyncRequestReadableStream, &AsyncRequestReadableStream::read>(*this);
+    (void)AsyncReadableStream::eventClose
+        .addListener<AsyncRequestReadableStream, &AsyncRequestReadableStream::onCloseStopRequest>(*this);
 }
 
 template <typename AsyncReadRequest>
-void SC::AsyncRequestReadableStream<AsyncReadRequest>::onEndCloseDescriptor()
+void SC::AsyncRequestReadableStream<AsyncReadRequest>::onCloseStopRequest()
 {
-    Result res = Internal::closeDescriptor(request);
-    if (not res)
+    if (not request.isFree())
     {
-        emitError(res);
+        request.stop(*eventLoop);
     }
-}
-
-template <typename AsyncReadRequest>
-SC::Result SC::AsyncRequestReadableStream<AsyncReadRequest>::registerAutoCloseDescriptor(bool value)
-{
-    using Self = AsyncRequestReadableStream;
-    if (value)
-    {
-        return Result(eventEnd.addListener<Self, &Self::onEndCloseDescriptor>(*this));
-    }
-    else
+    if (autoCloseDescriptor)
     {
-        return Result(eventEnd.removeListener<Self, &Self::onEndCloseDescriptor>(*this));
+        Result res = Internal::closeDescriptor(request);
+        if (not res)
+        {
+            emitError(res);
+        }
     }
 }
 
@@ -89,10 +84,10 @@ template <typename AsyncReadRequest>
 void SC::AsyncRequestReadableStream<AsyncReadRequest>::afterRead(typename AsyncReadRequest::Result& result,
                                                                  AsyncBufferView::ID                bufferID)
 {
-    SC_ASSERT_RELEASE(request.isFree());
     Span<char> data;
     if (result.get(data))
     {
+        SC_ASSERT_RELEASE(request.isFree());
         if (Internal::isEnded(result))
         {
             getBuffersPool().unrefBuffer(bufferID);
@@ -157,29 +152,24 @@ template <typename AsyncWriteRequest>
 SC::AsyncRequestWritableStream<AsyncWriteRequest>::AsyncRequestWritableStream()
 {
     AsyncWritableStream::asyncWrite.bind<AsyncRequestWritableStream, &AsyncRequestWritableStream::write>(*this);
+    (void)AsyncWritableStream::eventFinish
+        .addListener<AsyncRequestWritableStream, &AsyncRequestWritableStream::onFinishStopRequest>(*this);
 }
 
 template <typename AsyncWriteRequest>
-void SC::AsyncRequestWritableStream<AsyncWriteRequest>::onEndCloseDescriptor()
+void SC::AsyncRequestWritableStream<AsyncWriteRequest>::onFinishStopRequest()
 {
-    Result res = Internal::closeDescriptor(request);
-    if (not res)
+    if (not request.isFree())
     {
-        emitError(res);
+        request.stop(*eventLoop);
     }
-}
-
-template <typename AsyncWriteRequest>
-SC::Result SC::AsyncRequestWritableStream<AsyncWriteRequest>::registerAutoCloseDescriptor(bool value)
-{
-    using Self = AsyncRequestWritableStream;
-    if (value)
+    if (autoCloseDescriptor)
     {
-        return Result(eventFinish.addListener<Self, &Self::onEndCloseDescriptor>(*this));
-    }
-    else
-    {
-        return Result(eventFinish.removeListener<Self, &Self::onEndCloseDescriptor>(*this));
+        Result res = Internal::closeDescriptor(request);
+        if (not res)
+        {
+            emitError(res);
+        }
     }
 }
 

Libraries/AsyncStreams/AsyncRequestStreams.h

@@ -13,18 +13,20 @@ struct AsyncRequestReadableStream : public AsyncReadableStream
 {
     AsyncRequestReadableStream();
 
-    /// @brief Registers or unregisters a listener to AsyncReadableStream::eventEnd to close descriptor
-    Result registerAutoCloseDescriptor(bool value);
+    /// @brief Automatically closes descriptor during read stream close event
+    void setAutoCloseDescriptor(bool value) { autoCloseDescriptor = value; }
 
     AsyncRequestType request; /// AsyncFileRead / AsyncFileWrite / AsyncSocketReceive / AsyncSocketSend
   protected:
-    AsyncEventLoop* eventLoop = nullptr;
     struct Internal;
+    AsyncEventLoop* eventLoop = nullptr;
+
+    bool autoCloseDescriptor = false;
 
     Result read();
 
     void afterRead(typename AsyncRequestType::Result& result, AsyncBufferView::ID bufferID);
-    void onEndCloseDescriptor();
+    void onCloseStopRequest();
 };
 
 template <typename AsyncRequestType>
@@ -36,20 +38,22 @@ struct AsyncRequestWritableStream : public AsyncWritableStream
     Result init(AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& eventLoop,
                 const DescriptorType& descriptor);
 
-    /// @brief Registers or unregisters a listener to AsyncWritableStream::eventFinish to close descriptor
-    Result registerAutoCloseDescriptor(bool value);
+    /// @brief Automatically closes descriptor during write stream finish event
+    void setAutoCloseDescriptor(bool value) { autoCloseDescriptor = value; }
 
     AsyncRequestType request; /// AsyncFileRead / AsyncFileWrite / AsyncSocketReceive / AsyncSocketSend
 
   protected:
-    AsyncEventLoop* eventLoop = nullptr;
     struct Internal;
+    AsyncEventLoop* eventLoop = nullptr;
+
+    bool autoCloseDescriptor = false;
 
     Function<void(AsyncBufferView::ID)> callback;
 
     Result write(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb);
 
-    void onEndCloseDescriptor();
+    void onFinishStopRequest();
 };
 
 /// @brief Uses an SC::AsyncFileRead to stream data from a file

Tests/Libraries/AsyncStreams/AsyncRequestStreamsTest.cpp

@@ -379,8 +379,8 @@ void SC::AsyncRequestStreamsTest::fileCompressRemote(AsyncEventLoop& eventLoop,
     {
         SC_TEST_EXPECT(writeSideStream.request.executeOn(writeStreamTask, streamPool));
     }
-    SC_TEST_EXPECT(writeSideStream.registerAutoCloseDescriptor(true));
-    writeSide.detach(); // Taken care by registerAutoCloseDescriptor(true)
+    writeSideStream.setAutoCloseDescriptor(true);
+    writeSide.detach(); // Taken care by setAutoCloseDescriptor(true)
 
     // Create Readable Socket Stream
     READABLE_TYPE                readSideStream;
@@ -393,8 +393,8 @@ void SC::AsyncRequestStreamsTest::fileCompressRemote(AsyncEventLoop& eventLoop,
     {
         SC_TEST_EXPECT(readSideStream.request.executeOn(readStreamTask, streamPool));
     }
-    SC_TEST_EXPECT(readSideStream.registerAutoCloseDescriptor(true));
-    readSide.detach(); // Taken care by registerAutoCloseDescriptor(true)
+    readSideStream.setAutoCloseDescriptor(true);
+    readSide.detach(); // Taken care by setAutoCloseDescriptor(true)
     (void)readSideStream.eventError.addListener([this](Result res) { SC_TEST_EXPECT(res); });
 
     AsyncWritableStream::Request writeFileRequests[numberOfBuffers2 + 1];