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