src/shared/channel.cpp

   1 /*
   2  * sensord
   3  *
   4  * Copyright (c) 2017 Samsung Electronics Co., Ltd.
   5  *
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  * http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  */
  19
  20 #include "channel.h"
  21
  22 #include <stdint.h>
  23 #include <unistd.h>
  24 #include <memory>
  25 #include <algorithm>
  26
  27 #include "sensor_log.h"
  28 #include "channel_event_handler.h"
  29
  30 #define SYSTEMD_SOCK_BUF_SIZE (128*1024)
  31
  32 using namespace ipc;
  33
  34 class send_event_handler : public event_handler
  35 {
  36 public:
  37         send_event_handler(channel *ch, std::shared_ptr<message> msg)
  38         : m_ch(ch)
  39         , m_msg(msg)
  40         { }
  41
  42         bool handle(int fd, event_condition condition)
  43         {
  44                 if (!m_ch) {
  45                         return false;
  46                 }
  47
  48                 m_ch->remove_pending_event_id(m_event_id);
  49
  50                 if (!m_ch->is_connected()) {
  51                         return false;
  52                 }
  53
  54                 if (condition & (EVENT_IN | EVENT_HUP)) {
  55                         return false;
  56                 }
  57
  58                 if (!m_ch->send_sync(*m_msg)) {
  59                         return false;
  60                 }
  61
  62                 return false;
  63         }
  64
  65 private:
  66         channel *m_ch;
  67         std::shared_ptr<message> m_msg;
  68 };
  69
  70 class read_event_handler : public event_handler
  71 {
  72 public:
  73         read_event_handler(channel *ch)
  74         : m_ch(ch)
  75         { }
  76
  77         bool handle(int fd, event_condition condition)
  78         {
  79                 if (!m_ch) {
  80                         return false;
  81                 }
  82
  83                 m_ch->remove_pending_event_id(m_event_id);
  84
  85                 if (!m_ch->is_connected()) {
  86                         return false;
  87                 }
  88
  89                 if (condition & (EVENT_OUT | EVENT_HUP)) {
  90                         return false;
  91                 }
  92
  93                 message msg;
  94                 if (!m_ch->read_sync(msg, false)) {
  95                         return false;
  96                 }
  97
  98                 return false;
  99         }
 100
 101 private:
 102         channel *m_ch;
 103 };
 104
 105 channel::channel(socket *sock)
 106 : m_fd(sock->get_fd())
 107 , m_event_id(0)
 108 , m_socket(sock)
 109 , m_handler(NULL)
 110 , m_loop(NULL)
 111 , m_connected(false)
 112 {
 113         _D("Create[%p]", this);
 114 }
 115
 116 channel::~channel()
 117 {
 118         _D("Destroy[%p]", this);
 119         if (is_connected()) {
 120                 disconnect();
 121         }
 122 }
 123
 124 uint64_t channel::bind(void)
 125 {
 126         retv_if(!m_loop, 0);
 127         m_event_id = m_loop->add_event(m_socket->get_fd(),
 128                         (EVENT_IN | EVENT_HUP | EVENT_NVAL),
 129                         dynamic_cast<channel_event_handler *>(m_handler));
 130
 131         _D("Bind channel[%p] : handler[%p] event_id[%llu]", this, m_handler, m_event_id);
 132         return m_event_id;
 133 }
 134
 135 uint64_t channel::bind(channel_handler *handler, event_loop *loop, bool loop_bind)
 136 {
 137         m_handler = handler;
 138         m_loop = loop;
 139         m_connected.store(true);
 140
 141         if (m_handler)
 142                 m_handler->connected(this);
 143
 144         if (loop_bind)
 145                 bind();
 146
 147         return m_event_id;
 148 }
 149
 150 uint64_t channel::connect(channel_handler *handler, event_loop *loop, bool loop_bind)
 151 {
 152         if (!m_socket->connect())
 153                 return false;
 154
 155         bind(handler, loop, loop_bind);
 156
 157         _D("Connect channel[%p] : event id[%llu]", this, m_event_id);
 158         return m_event_id;
 159 }
 160
 161 void channel::disconnect(void)
 162 {
 163         AUTOLOCK(m_cmutex);
 164         if (!is_connected()) {
 165                 _D("Channel[%p] is not connected", this);
 166                 return;
 167         }
 168
 169         m_connected.store(false);
 170
 171         _D("Disconnect channel[%p]", this);
 172
 173         if (m_handler) {
 174                 _D("Disconnect channel[%p] handler[%p]", this, m_handler);
 175                 m_handler->disconnected(this);
 176                 m_handler = NULL;
 177         }
 178
 179         if (m_loop) {
 180                 for(auto id : m_pending_event_id) {
 181                         _D("Remove channel[%p] pending event id[%llu]", this, id);
 182                         m_loop->remove_event(id, true);
 183                 }
 184                 _D("Remove channel[%p] event[%llu]",this, m_event_id);
 185                 m_loop->remove_event(m_event_id, true);
 186                 m_event_id = 0;
 187         }
 188
 189         if (m_socket) {
 190                 _D("Release channel[%p] socket[%d]", this, m_socket->get_fd());
 191                 delete m_socket;
 192                 m_socket = NULL;
 193         }
 194         _D("Channel[%p] is disconnected", this);
 195 }
 196
 197 bool channel::send(std::shared_ptr<message> msg)
 198 {
 199         int retry_cnt = 0;
 200         int cur_buffer_size = 0;
 201
 202         retv_if(!m_loop, false);
 203
 204         while (retry_cnt < 3) {
 205                 cur_buffer_size = m_socket->get_current_buffer_size();
 206                 if (cur_buffer_size <= SYSTEMD_SOCK_BUF_SIZE)
 207                         break;
 208                 usleep(3000);
 209                 retry_cnt++;
 210         }
 211         retvm_if(retry_cnt >= 3, false, "Socket buffer[%d] is exceeded", cur_buffer_size);
 212
 213         send_event_handler *handler = new(std::nothrow) send_event_handler(this, msg);
 214         retvm_if(!handler, false, "Failed to allocate memory");
 215
 216         uint64_t event_id = m_loop->add_event(m_socket->get_fd(), (EVENT_OUT | EVENT_HUP | EVENT_NVAL), handler);
 217         if (event_id == 0) {
 218                 _D("Failed to add send event handler");
 219                 delete handler;
 220                 return false;
 221         }
 222
 223         m_pending_event_id.push_back(event_id);
 224         return true;
 225 }
 226
 227 bool channel::send_sync(message &msg)
 228 {
 229         AUTOLOCK(m_cmutex);
 230         if (!is_connected()) {
 231                 _D("Channel is not connected");
 232                 return false;
 233         }
 234
 235         retvm_if(msg.size() >= MAX_MSG_CAPACITY, true, "Invaild message size[%u]", msg.size());
 236
 237         ssize_t size = 0;
 238         char *buf = msg.body();
 239
 240         /* header */
 241         size = m_socket->send(reinterpret_cast<void *>(msg.header()),
 242             sizeof(message_header), true);
 243         retvm_if(size <= 0, false, "Failed to send header");
 244
 245         /* if body size is zero, skip to send body message */
 246         retv_if(msg.size() == 0, true);
 247
 248         /* body */
 249         size = m_socket->send(buf, msg.size(), true);
 250         retvm_if(size <= 0, false, "Failed to send body");
 251
 252         return true;
 253 }
 254
 255 bool channel::read(void)
 256 {
 257         retv_if(!m_loop, false);
 258
 259         read_event_handler *handler = new(std::nothrow) read_event_handler(this);
 260         retvm_if(!handler, false, "Failed to allocate memory");
 261
 262         uint64_t event_id = m_loop->add_event(m_socket->get_fd(), (EVENT_IN | EVENT_HUP | EVENT_NVAL), handler);
 263         if (event_id == 0) {
 264                 _D("Failed to add read event handler");
 265                 delete handler;
 266                 return false;
 267         }
 268
 269         m_pending_event_id.push_back(event_id);
 270         return true;
 271 }
 272
 273 bool channel::read_sync(message &msg, bool select)
 274 {
 275         AUTOLOCK(m_cmutex);
 276         if (!is_connected()) {
 277                 _D("Channel is not connected");
 278                 return false;
 279         }
 280
 281         message_header header;
 282         ssize_t size = 0;
 283         char buf[MAX_MSG_CAPACITY];
 284
 285         /* header */
 286         size = m_socket->recv(&header, sizeof(message_header), select);
 287         retv_if(size <= 0, false);
 288
 289         /* check error from header */
 290         if (m_handler && header.err != 0) {
 291                 m_handler->error_caught(this, header.err);
 292                 msg.header()->err = header.err;
 293                 return true;
 294         }
 295
 296         /* body */
 297         if (header.length >= MAX_MSG_CAPACITY) {
 298                 _E("header.length error %u", header.length);
 299                 return false;
 300         }
 301
 302         if (header.length > 0) {
 303                 size = m_socket->recv(&buf, header.length, select);
 304                 retv_if(size <= 0, false);
 305         }
 306
 307         buf[header.length] = '\0';
 308         msg.enclose(reinterpret_cast<const void *>(buf), header.length);
 309         msg.set_type(header.type);
 310         msg.header()->err = header.err;
 311
 312         if (m_handler)
 313                 m_handler->read(this, msg);
 314
 315         return true;
 316 }
 317
 318 bool channel::is_connected(void)
 319 {
 320         return m_connected.load();
 321 }
 322
 323 bool channel::set_option(int type, int value)
 324 {
 325         switch (type) {
 326         case SO_SNDBUF:
 327                 m_socket->set_buffer_size(type, value);
 328                 break;
 329         case SO_RCVBUF:
 330                 m_socket->set_buffer_size(type, value);
 331                 break;
 332         default:
 333                 break;
 334         }
 335
 336         return true;
 337 }
 338
 339 bool channel::get_option(int type, int &value) const
 340 {
 341         switch (type) {
 342         case 0:
 343                 value = m_socket->get_current_buffer_size();
 344                 break;
 345         case SO_SNDBUF:
 346                 value = m_socket->get_buffer_size(type);
 347                 break;
 348         case SO_RCVBUF:
 349                 value = m_socket->get_buffer_size(type);
 350                 break;
 351         default:
 352                 break;
 353         }
 354
 355         return true;
 356 }
 357
 358 int channel::get_fd(void) const
 359 {
 360         return m_fd;
 361 }
 362
 363 void channel::remove_pending_event_id(uint64_t id)
 364 {
 365         auto it = std::find(m_pending_event_id.begin(), m_pending_event_id.end(), id);
 366         if (it != m_pending_event_id.end()) {
 367                 m_pending_event_id.erase(it);
 368         }
 369 }