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