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