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