modules/socket/src/waitable_input_output_execution_context_support.cpp

   1 /*
   2  * Copyright (c) 2011 Samsung Electronics Co., Ltd All Rights Reserved
   3  *
   4  *    Licensed under the Apache License, Version 2.0 (the "License");
   5  *    you may not use this file except in compliance with the License.
   6  *    You may obtain a copy of the License at
   7  *
   8  *        http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  *    Unless required by applicable law or agreed to in writing, software
  11  *    distributed under the License is distributed on an "AS IS" BASIS,
  12  *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  *    See the License for the specific language governing permissions and
  14  *    limitations under the License.
  15  */
  16 /*
  17  * @file        waitable_input_output_execution_context_support.cpp
  18  * @author      Przemyslaw Dobrowolski (p.dobrowolsk@samsung.com)
  19  * @version     1.0
  20  * @brief       This file is the implementation file of waitable input-output execution context support
  21  */
  22 #include <stddef.h>
  23 #include <dpl/socket/waitable_input_output_execution_context_support.h>
  24 #include <dpl/scoped_array.h>
  25 #include <dpl/socket/abstract_socket.h> // FIXME: Remove !!!
  26 #include <dpl/log/log.h>
  27 #include <dpl/assert.h>
  28
  29 namespace DPL
  30 {
  31 namespace Socket
  32 {
  33
  34 namespace // anonymous
  35 {
  36 const size_t DEFAULT_READ_SIZE = 2048;
  37 } // namespace anonymous
  38
  39 WaitableInputOutputExecutionContextSupport::WaitableInputOutputExecutionContextSupport()
  40     : m_opened(false),
  41       m_waitableInputOutput(NULL),
  42       m_hasReadWatch(false),
  43       m_hasWriteWatch(false)
  44 {
  45 }
  46
  47 WaitableInputOutputExecutionContextSupport::~WaitableInputOutputExecutionContextSupport()
  48 {
  49     // Ensure support is closed
  50     Close();
  51 }
  52
  53 void WaitableInputOutputExecutionContextSupport::Open(AbstractWaitableInputOutput *inputOutput)
  54 {
  55     if (m_opened)
  56         Throw(Exception::AlreadyOpened);
  57
  58     LogPedantic("Opening waitable input-output execution context support...");
  59
  60     // Save IO handle
  61     m_waitableInputOutput = inputOutput;
  62
  63     // Register read watch
  64     Assert(m_hasReadWatch == false);
  65
  66     AddReadWatch();
  67     m_hasReadWatch = true;
  68
  69     // Done
  70     m_opened = true;
  71
  72     LogPedantic("Waitable input-output execution context support opened");
  73 }
  74
  75 void WaitableInputOutputExecutionContextSupport::Close()
  76 {
  77     if (!m_opened)
  78         return;
  79
  80     LogPedantic("Closing waitable input-output execution context support...");
  81
  82     // Remove read and write watches
  83     CheckedRemoveReadWriteWatch();
  84
  85     // Set proper state
  86     m_opened = false;
  87
  88     LogPedantic("Waitable input-output execution context support closed");
  89 }
  90
  91 void WaitableInputOutputExecutionContextSupport::AddReadWatch()
  92 {
  93     WaitableHandleWatchSupport::InheritedContext()->AddWaitableHandleWatch(this, m_waitableInputOutput->WaitableReadHandle(), WaitMode::Read);
  94 }
  95
  96 void WaitableInputOutputExecutionContextSupport::RemoveReadWatch()
  97 {
  98     WaitableHandleWatchSupport::InheritedContext()->RemoveWaitableHandleWatch(this, m_waitableInputOutput->WaitableReadHandle(), WaitMode::Read);
  99 }
 100
 101 void WaitableInputOutputExecutionContextSupport::AddWriteWatch()
 102 {
 103     WaitableHandleWatchSupport::InheritedContext()->AddWaitableHandleWatch(this, m_waitableInputOutput->WaitableWriteHandle(), WaitMode::Write);
 104 }
 105
 106 void WaitableInputOutputExecutionContextSupport::RemoveWriteWatch()
 107 {
 108     WaitableHandleWatchSupport::InheritedContext()->RemoveWaitableHandleWatch(this, m_waitableInputOutput->WaitableWriteHandle(), WaitMode::Write);
 109 }
 110
 111 void WaitableInputOutputExecutionContextSupport::CheckedRemoveReadWatch()
 112 {
 113     if (!m_hasReadWatch)
 114         return;
 115
 116     RemoveReadWatch();
 117     m_hasReadWatch = false;
 118 }
 119
 120 void WaitableInputOutputExecutionContextSupport::CheckedRemoveWriteWatch()
 121 {
 122     if (!m_hasWriteWatch)
 123         return;
 124
 125     RemoveWriteWatch();
 126     m_hasWriteWatch = false;
 127 }
 128
 129 void WaitableInputOutputExecutionContextSupport::CheckedRemoveReadWriteWatch()
 130 {
 131     // Remove read watch if any
 132     CheckedRemoveReadWatch();
 133
 134     // Remove write watch if any
 135     CheckedRemoveWriteWatch();
 136 }
 137
 138 void WaitableInputOutputExecutionContextSupport::OnWaitableHandleEvent(WaitableHandle waitableHandle, WaitMode::Type mode)
 139 {
 140     (void)waitableHandle;
 141
 142     switch (mode)
 143     {
 144         case WaitMode::Read:
 145             LogPedantic("Read event occurred");
 146
 147             // Read and parse bytes
 148             ReadInput();
 149
 150             // Done
 151             break;
 152
 153         case WaitMode::Write:
 154             LogPedantic("Write event occurred");
 155
 156             // Push bytes and unregister from write event
 157             FeedOutput();
 158
 159             // Unregister write watch only if no more data is available
 160             if (m_outputStream.Empty())
 161             {
 162                 Assert(m_hasWriteWatch == true);
 163                 CheckedRemoveWriteWatch();
 164             }
 165
 166             // Done
 167             break;
 168
 169         default:
 170             Assert(0);
 171             break;
 172     }
 173 }
 174
 175 void WaitableInputOutputExecutionContextSupport::ReadInput()
 176 {
 177     LogPedantic("Reading input bytes");
 178
 179     Try
 180     {
 181         BinaryQueueAutoPtr inputBuffer = m_waitableInputOutput->Read(DEFAULT_READ_SIZE);
 182
 183         if (inputBuffer.get() == NULL)
 184         {
 185             // No data, should not occur
 186             LogPedantic("WARNING: Spontaneous ReadSocket occurred");
 187             return;
 188         }
 189
 190         if (inputBuffer->Empty())
 191         {
 192             // Connection was closed
 193             OnInputStreamClosed();
 194
 195             // Unregister from further event insisting
 196             Assert(m_hasReadWatch == true);
 197             CheckedRemoveReadWriteWatch();
 198
 199             // Set proper state
 200             m_opened = false;
 201
 202             // Done
 203             return;
 204         }
 205
 206         LogPedantic("Read " << inputBuffer->Size() << " input bytes");
 207
 208         // Append all read data
 209         m_inputStream.AppendMoveFrom(*inputBuffer);
 210     }
 211     Catch (AbstractSocket::Exception::ConnectionBroken) // FIXME: Inproper exception abstraction !!!
 212     {
 213         // Some errors occurred while feeding abstract IO
 214         // Interpret connection broken errors, and pass futher other ones
 215         LogPedantic("Abstract IO connection was broken during read");
 216
 217         // Signal broken connection
 218         OnInputStreamBroken();
 219
 220         // Unregister from further event insisting
 221         Assert(m_hasReadWatch == true);
 222         CheckedRemoveReadWriteWatch();
 223
 224         // Set proper state
 225         m_opened = false;
 226
 227         // Do not continue
 228         return;
 229     }
 230
 231     // Interpret data
 232     OnInputStreamRead();
 233 }
 234
 235 void WaitableInputOutputExecutionContextSupport::FeedOutput()
 236 {
 237     if (!m_opened)
 238         Throw(Exception::NotOpened);
 239
 240     // Anything to feed ?
 241     if (m_outputStream.Empty())
 242         return;
 243
 244     // OK to feed output
 245     LogPedantic("Feeding output");
 246
 247     Try
 248     {
 249         // Try to write some bytes
 250         size_t bytes = m_waitableInputOutput->Write(m_outputStream, m_outputStream.Size());
 251
 252         if (bytes < m_outputStream.Size())
 253         {
 254             // Start exhaustive output feeding if it is blocked and not already started
 255             if (!m_hasWriteWatch)
 256             {
 257                 AddWriteWatch();
 258                 m_hasWriteWatch = true;
 259
 260                 LogPedantic("Started exhaustive output feeding");
 261             }
 262         }
 263
 264         // Some bytes were written, consume them
 265         m_outputStream.Consume(bytes);
 266     }
 267     Catch (AbstractSocket::Exception::ConnectionBroken) // FIXME: Inproper exception abstraction !!!
 268     {
 269         // Some errors occurred while feeding abstract IO
 270         // Interpret connection broken errors, and pass futher other ones
 271         LogPedantic("Abstract IO connection was broken during write");
 272
 273         // Signal broken connection
 274         OnInputStreamBroken();
 275
 276         // Unregister from further event insisting
 277         Assert(m_hasReadWatch == true);
 278         CheckedRemoveReadWriteWatch();
 279
 280         // Set proper state
 281         m_opened = false;
 282
 283         // Do not continue
 284         return;
 285     }
 286 }
 287
 288 }
 289 } // namespace DPL