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
  26 #include "sensor_log.h"
  27 #include "channel_event_handler.h"
  28
  29 #define SYSTEMD_SOCK_BUF_SIZE (128*1024)
  30
  31 using namespace ipc;
  32
  33 class send_event_handler : public event_handler
  34 {
  35 public:
  36         send_event_handler(channel *ch, std::shared_ptr<message> msg)
  37         : m_ch(ch)
  38         , m_msg(msg)
  39         { }
  40
  41         bool handle(int fd, event_condition condition)
  42         {
  43                 if (!m_ch || !m_ch->is_connected())
  44                         return false;
  45
  46                 if (condition & (EVENT_IN | EVENT_HUP))
  47                         return false;
  48
  49                 if (!m_ch->send_sync(*m_msg))
  50                         return false;
  51
  52                 return false;
  53         }
  54
  55 private:
  56         channel *m_ch;
  57         std::shared_ptr<message> m_msg;
  58 };
  59
  60 class read_event_handler : public event_handler
  61 {
  62 public:
  63         read_event_handler(channel *ch)
  64         : m_ch(ch)
  65         { }
  66
  67         bool handle(int fd, event_condition condition)
  68         {
  69                 message msg;
  70
  71                 if (!m_ch || !m_ch->is_connected())
  72                         return false;
  73
  74                 if (condition & (EVENT_OUT | EVENT_HUP))
  75                         return false;
  76
  77                 if (!m_ch->read_sync(msg, false))
  78                         return false;
  79
  80                 return false;
  81         }
  82
  83 private:
  84         channel *m_ch;
  85 };
  86
  87 channel::channel(socket *sock)
  88 : m_fd(sock->get_fd())
  89 , m_event_id(0)
  90 , m_socket(sock)
  91 , m_handler(NULL)
  92 , m_loop(NULL)
  93 , m_connected(false)
  94 {
  95         _D("Created");
  96 }
  97
  98 channel::~channel()
  99 {
 100         _D("Destroyed[%llu]", m_event_id);
 101         disconnect();
 102 }
 103
 104 uint64_t channel::bind(void)
 105 {
 106         retv_if(!m_loop, 0);
 107         m_event_id = m_loop->add_event(m_socket->get_fd(),
 108                         (EVENT_IN | EVENT_HUP | EVENT_NVAL),
 109                         dynamic_cast<channel_event_handler *>(m_handler));
 110
 111         _D("Bound[%llu]", m_event_id);
 112         return m_event_id;
 113 }
 114
 115 uint64_t channel::bind(channel_handler *handler, event_loop *loop, bool loop_bind)
 116 {
 117         m_handler = handler;
 118         m_loop = loop;
 119         m_connected.store(true);
 120
 121         if (m_handler)
 122                 m_handler->connected(this);
 123
 124         if (loop_bind)
 125                 bind();
 126
 127         return m_event_id;
 128 }
 129
 130 uint64_t channel::connect(channel_handler *handler, event_loop *loop, bool loop_bind)
 131 {
 132         if (!m_socket->connect())
 133                 return false;
 134
 135         bind(handler, loop, loop_bind);
 136
 137         _D("Connected[%llu]", m_event_id);
 138         return m_event_id;
 139 }
 140
 141 void channel::disconnect(void)
 142 {
 143         if (!is_connected()) {
 144                 _D("Channel is not connected");
 145                 return;
 146         }
 147
 148         m_connected.store(false);
 149
 150         _D("Disconnecting..[%llu]", m_event_id);
 151
 152         if (m_handler) {
 153                 m_handler->disconnected(this);
 154                 m_handler = NULL;
 155         }
 156
 157         if (m_loop) {
 158                 _D("Remove event[%llu]", m_event_id);
 159                 m_loop->remove_event(m_event_id, true);
 160                 m_loop = NULL;
 161                 m_event_id = 0;
 162         }
 163
 164         if (m_socket) {
 165                 _D("Release socket[%d]", m_socket->get_fd());
 166                 delete m_socket;
 167                 m_socket = NULL;
 168         }
 169
 170         _D("Disconnected");
 171 }
 172
 173 bool channel::send(std::shared_ptr<message> msg)
 174 {
 175         int retry_cnt = 0;
 176         int cur_buffer_size = 0;
 177
 178         retv_if(!m_loop, false);
 179
 180         while (retry_cnt < 3) {
 181                 cur_buffer_size = m_socket->get_current_buffer_size();
 182                 if (cur_buffer_size <= SYSTEMD_SOCK_BUF_SIZE)
 183                         break;
 184                 usleep(3000);
 185                 retry_cnt++;
 186         }
 187         retvm_if(retry_cnt >= 3, false, "Socket buffer[%d] is exceeded", cur_buffer_size);
 188
 189         send_event_handler *handler = new(std::nothrow) send_event_handler(this, msg);
 190         retvm_if(!handler, false, "Failed to allocate memory");
 191
 192         if (m_loop->add_event(m_socket->get_fd(), (EVENT_OUT | EVENT_HUP | EVENT_NVAL) , handler) == 0) {
 193                 _D("Failed to add send event handler");
 194                 delete handler;
 195                 return false;
 196         }
 197
 198         return true;
 199 }
 200
 201 bool channel::send_sync(message &msg)
 202 {
 203         retvm_if(msg.size() >= MAX_MSG_CAPACITY, true, "Invaild message size[%u]", msg.size());
 204
 205         ssize_t size = 0;
 206         char *buf = msg.body();
 207
 208         /* header */
 209         size = m_socket->send(reinterpret_cast<void *>(msg.header()),
 210             sizeof(message_header), true);
 211         retvm_if(size <= 0, false, "Failed to send header");
 212
 213         /* if body size is zero, skip to send body message */
 214         retv_if(msg.size() == 0, true);
 215
 216         /* body */
 217         size = m_socket->send(buf, msg.size(), true);
 218         retvm_if(size <= 0, false, "Failed to send body");
 219
 220         return true;
 221 }
 222
 223 bool channel::read(void)
 224 {
 225         retv_if(!m_loop, false);
 226
 227         read_event_handler *handler = new(std::nothrow) read_event_handler(this);
 228         retvm_if(!handler, false, "Failed to allocate memory");
 229
 230         if (m_loop->add_event(m_socket->get_fd(), (EVENT_IN | EVENT_HUP | EVENT_NVAL), handler) == 0) {
 231                 _D("Failed to add read event handler");
 232                 delete handler;
 233                 return false;
 234         }
 235
 236         return true;
 237 }
 238
 239 bool channel::read_sync(message &msg, bool select)
 240 {
 241         message_header header;
 242         ssize_t size = 0;
 243         char buf[MAX_MSG_CAPACITY];
 244
 245         /* header */
 246         size = m_socket->recv(&header, sizeof(message_header), select);
 247         retv_if(size <= 0, false);
 248
 249         /* check error from header */
 250         if (m_handler && header.err != 0) {
 251                 m_handler->error_caught(this, header.err);
 252                 msg.header()->err = header.err;
 253                 return true;
 254         }
 255
 256         /* body */
 257         if (header.length >= MAX_MSG_CAPACITY) {
 258                 _E("header.length error %u", header.length);
 259                 return false;
 260         }
 261
 262         if (header.length > 0) {
 263                 size = m_socket->recv(&buf, header.length, select);
 264                 retv_if(size <= 0, false);
 265         }
 266
 267         buf[header.length] = '\0';
 268         msg.enclose(reinterpret_cast<const void *>(buf), header.length);
 269         msg.set_type(header.type);
 270         msg.header()->err = header.err;
 271
 272         if (m_handler)
 273                 m_handler->read(this, msg);
 274
 275         return true;
 276 }
 277
 278 bool channel::is_connected(void)
 279 {
 280         return m_connected.load();
 281 }
 282
 283 bool channel::set_option(int type, int value)
 284 {
 285         switch (type) {
 286         case SO_SNDBUF:
 287                 m_socket->set_buffer_size(type, value);
 288                 break;
 289         case SO_RCVBUF:
 290                 m_socket->set_buffer_size(type, value);
 291                 break;
 292         default:
 293                 break;
 294         }
 295
 296         return true;
 297 }
 298
 299 bool channel::get_option(int type, int &value) const
 300 {
 301         switch (type) {
 302         case 0:
 303                 value = m_socket->get_current_buffer_size();
 304                 break;
 305         case SO_SNDBUF:
 306                 value = m_socket->get_buffer_size(type);
 307                 break;
 308         case SO_RCVBUF:
 309                 value = m_socket->get_buffer_size(type);
 310                 break;
 311         default:
 312                 break;
 313         }
 314
 315         return true;
 316 }
 317
 318 int channel::get_fd(void) const
 319 {
 320         return m_fd;
 321 }