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