modules_wearable/rpc/src/generic_rpc_connection.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        generic_rpc_connection.cpp
  18  * @author      Przemyslaw Dobrowolski (p.dobrowolsk@samsung.com)
  19  * @version     1.0
  20  * @brief       This file is the implementation file of generic RPC connection
  21  */
  22 #include <stddef.h>
  23 #include <dpl/rpc/generic_rpc_connection.h>
  24 #include <dpl/scoped_array.h>
  25 #include <dpl/log/log.h>
  26 #include <dpl/aligned.h>
  27 #include <stdexcept>
  28
  29 namespace DPL {
  30 namespace RPC {
  31 namespace // anonymous
  32 {
  33 namespace Protocol {
  34 // Packet definitions
  35 enum PacketType
  36 {
  37     PacketType_AsyncCall,
  38     PacketType_PingPong
  39 };
  40
  41 struct Header
  42 {
  43     unsigned short size;
  44     unsigned short type;
  45 } DPL_ALIGNED(1);
  46
  47 struct AsyncCall :
  48     public Header
  49 {
  50     unsigned char data[1];
  51 } DPL_ALIGNED(1);
  52 } // namespace Protocol
  53 } // namespace anonymous
  54
  55 GenericRPCConnection::GenericRPCConnection(
  56     AbstractWaitableInputOutput *inputOutput) :
  57     m_inputOutput(inputOutput)
  58 {
  59     LogPedantic("Opening generic RPC...");
  60     WaitableInputOutputExecutionContextSupport::Open(inputOutput);
  61     LogPedantic("Generic RPC opened");
  62 }
  63
  64 GenericRPCConnection::~GenericRPCConnection()
  65 {
  66     // Ensure RPC is closed
  67     LogPedantic("Closing generic RPC...");
  68     WaitableInputOutputExecutionContextSupport::Close();
  69     LogPedantic("Generic RPC closed");
  70 }
  71
  72 void GenericRPCConnection::AsyncCall(const RPCFunction &function)
  73 {
  74     LogPedantic("Executing async call");
  75
  76     // Create binary call
  77     BinaryQueue serializedCall = function.Serialize();
  78
  79     // Append buffers
  80     Protocol::AsyncCall call;
  81     call.size = static_cast<unsigned short>(serializedCall.Size());
  82     call.type = Protocol::PacketType_AsyncCall;
  83
  84     m_outputStream.AppendCopy(&call, sizeof(Protocol::Header));
  85     m_outputStream.AppendMoveFrom(serializedCall);
  86
  87     // Try to feed output with data
  88     Try
  89     {
  90         FeedOutput();
  91     }
  92     Catch(WaitableInputOutputExecutionContextSupport::Exception::NotOpened)
  93     {
  94         // Error occurred while feeding
  95         ReThrow(AbstractRPCConnection::Exception::AsyncCallFailed);
  96     }
  97 }
  98
  99 void GenericRPCConnection::Ping()
 100 {
 101     LogPedantic("Executing ping call");
 102
 103     // Append buffers
 104     Protocol::AsyncCall call;
 105     call.size = 0;
 106     call.type = Protocol::PacketType_PingPong;
 107
 108     m_outputStream.AppendCopy(&call, sizeof(Protocol::Header));
 109
 110     // Try to feed output with data
 111     Try
 112     {
 113         FeedOutput();
 114     }
 115     Catch(WaitableInputOutputExecutionContextSupport::Exception::NotOpened)
 116     {
 117         // Error occurred while feeding
 118         ReThrow(AbstractRPCConnection::Exception::PingFailed);
 119     }
 120 }
 121
 122 void GenericRPCConnection::OnInputStreamRead()
 123 {
 124     LogPedantic("Interpreting " << m_inputStream.Size() << " bytes buffer");
 125
 126     // Enough bytes to read at least one header ?
 127     if (m_inputStream.Size() >= sizeof(Protocol::Header)) {
 128         // Begin consuming as much packets as it is possible
 129         while (m_inputStream.Size() >= sizeof(Protocol::Header)) {
 130             Protocol::Header header;
 131             m_inputStream.Flatten(&header, sizeof(header));
 132
 133             if (m_inputStream.Size() >= sizeof(Protocol::Header) +
 134                 header.size)
 135             {
 136                 LogPedantic("Will parse packet of type: " << header.type);
 137
 138                 // Allocate new packet (header + real packet data)
 139                 void *binaryPacket = malloc(
 140                         sizeof(Protocol::Header) + header.size);
 141
 142                 if (binaryPacket == NULL) {
 143                     throw std::bad_alloc();
 144                 }
 145
 146                 // Get it from stream
 147                 m_inputStream.FlattenConsume(
 148                     binaryPacket,
 149                     sizeof(Protocol::Header) +
 150                     header.size);
 151
 152                 // Parse specific packet
 153                 switch (header.type) {
 154                 case Protocol::PacketType_AsyncCall:
 155                 {
 156                     BinaryQueue call;
 157
 158                     // No need to delete packet data, we can use it
 159                     call.AppendUnmanaged(binaryPacket,
 160                                          sizeof(Protocol::Header) + header.size,
 161                                          &BinaryQueue::BufferDeleterFree,
 162                                          NULL);
 163
 164                     // ...but just remove protocol header
 165                     call.Consume(sizeof(Protocol::Header));
 166
 167                     LogPedantic(
 168                         "Async call of size: " << header.size <<
 169                         " parsed");
 170
 171                     // Call async call event listeners
 172                     DPL::Event::EventSupport<AbstractRPCConnectionEvents::
 173                                                  AsyncCallEvent>::
 174                         EmitEvent(AbstractRPCConnectionEvents::AsyncCallEvent(
 175                                       RPCFunction(call), EventSender(
 176                                           this)), DPL::Event::EmitMode::Queued);
 177                 }
 178                 break;
 179
 180                 case Protocol::PacketType_PingPong:
 181                 {
 182                     // Reply with ping/pong
 183                     Ping();
 184
 185                     // Do not need packet data
 186                     free(binaryPacket);
 187
 188                     LogPedantic("Ping pong replied");
 189                 }
 190                 break;
 191
 192                 default:
 193                     LogPedantic("Warning: Unknown packet type");
 194                     free(binaryPacket);
 195                     break;
 196                 }
 197             } else {
 198                 LogPedantic("Too few bytes to read packet");
 199                 break;
 200             }
 201         }
 202     } else {
 203         LogPedantic("Too few bytes to read header");
 204     }
 205 }
 206
 207 void GenericRPCConnection::OnInputStreamClosed()
 208 {
 209     // Emit closed event
 210     DPL::Event::EventSupport<AbstractRPCConnectionEvents::ConnectionClosedEvent>
 211         ::
 212         EmitEvent(AbstractRPCConnectionEvents::ConnectionClosedEvent(
 213                       EventSender(this)), DPL::Event::EmitMode::Queued);
 214 }
 215
 216 void GenericRPCConnection::OnInputStreamBroken()
 217 {
 218     // Emit broken event
 219     DPL::Event::EventSupport<AbstractRPCConnectionEvents::ConnectionBrokenEvent>
 220         ::
 221         EmitEvent(AbstractRPCConnectionEvents::ConnectionBrokenEvent(
 222                       EventSender(this)), DPL::Event::EmitMode::Queued);
 223 }
 224 }
 225 } // namespace DPL