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
  28 #define SYSTEMD_SOCK_BUF_SIZE 40000
  29
  30 using namespace ipc;
  31
  32 class send_event_handler : public event_handler
  33 {
  34 public:
  35         send_event_handler(channel *ch, message *msg)
  36         : m_ch(ch)
  37         , m_msg(msg)
  38         { }
  39
  40         bool handle(int fd, event_condition condition)
  41         {
  42                 if (!m_ch || !m_ch->is_connected())
  43                         return false;
  44
  45                 if (condition & (EVENT_IN | EVENT_HUP))
  46                         return false;
  47
  48                 if (!m_ch->send_sync(m_msg))
  49                         return false;
  50
  51                 if (m_msg)
  52                         m_msg->unref();
  53
  54                 return false;
  55         }
  56
  57 private:
  58         channel *m_ch;
  59         message *m_msg;
  60 };
  61
  62 class read_event_handler : public event_handler
  63 {
  64 public:
  65         read_event_handler(channel *ch)
  66         : m_ch(ch)
  67         { }
  68
  69         bool handle(int fd, event_condition condition)
  70         {
  71                 message msg;
  72
  73                 if (!m_ch || !m_ch->is_connected())
  74                         return false;
  75
  76                 if (condition & (EVENT_OUT | EVENT_HUP))
  77                         return false;
  78
  79                 if (!m_ch->read_sync(msg))
  80                         return false;
  81
  82                 return false;
  83         }
  84
  85 private:
  86         channel *m_ch;
  87 };
  88
  89 channel::channel(socket *sock)
  90 : m_fd(sock->get_fd())
  91 , m_event_id(0)
  92 , m_socket(sock)
  93 , m_handler(NULL)
  94 , m_loop(NULL)
  95 , m_connected(false)
  96 {
  97 }
  98
  99 channel::~channel()
 100 {
 101         /* disconnect() should not be called here */
 102 }
 103
 104 void channel::bind(channel_handler *handler, event_loop *loop)
 105 {
 106         m_handler = handler;
 107         m_loop = loop;
 108         m_connected.store(true);
 109
 110         if (m_handler)
 111                 m_handler->connected(this);
 112 }
 113
 114 bool channel::connect(channel_handler *handler, event_loop *loop)
 115 {
 116         if (!m_socket->connect())
 117                 return false;
 118
 119         bind(handler, loop);
 120         return true;
 121 }
 122
 123 void channel::disconnect(void)
 124 {
 125         ret_if(!is_connected());
 126         m_connected.store(false);
 127
 128         if (m_handler) {
 129                 m_handler->disconnected(this);
 130                 m_handler = NULL;
 131         }
 132
 133         if (m_loop) {
 134                 m_loop->remove_event(m_event_id, true);
 135                 m_loop = NULL;
 136                 m_event_id = 0;
 137         }
 138
 139         if (m_socket) {
 140                 delete m_socket;
 141                 m_socket = NULL;
 142         }
 143 }
 144
 145 bool channel::send(message *msg)
 146 {
 147         retv_if(!m_loop, false);
 148
 149         /* TODO: check buffer size(is there any linux api for this?) */
 150         int cur_buffer_size = m_socket->get_current_buffer_size();
 151         retvm_if(cur_buffer_size > SYSTEMD_SOCK_BUF_SIZE, false, "Failed to send data");
 152
 153         send_event_handler *handler = new(std::nothrow) send_event_handler(this, msg);
 154         retvm_if(!handler, false, "Failed to allocate memory");
 155
 156         msg->ref();
 157
 158         m_loop->add_event(m_socket->get_fd(),
 159                         (EVENT_OUT | EVENT_HUP | EVENT_NVAL) , handler);
 160
 161         return true;
 162 }
 163
 164 bool channel::send_sync(message *msg)
 165 {
 166         retv_if(!msg, false);
 167
 168         ssize_t size = 0;
 169         char *buf = msg->body();
 170
 171         /* header */
 172         size = m_socket->send(reinterpret_cast<void *>(msg->header()),
 173                                    sizeof(message_header), true);
 174         retv_if(size <= 0, false);
 175         retv_if(msg->size() <= 0, true);
 176
 177         /* body */
 178         size = m_socket->send(buf, msg->size(), true);
 179         retv_if(size <= 0, false);
 180
 181         return true;
 182 }
 183
 184 bool channel::read(void)
 185 {
 186         retv_if(!m_loop, false);
 187
 188         read_event_handler *handler = new(std::nothrow) read_event_handler(this);
 189         retvm_if(!handler, false, "Failed to allocate memory");
 190
 191         m_loop->add_event(m_socket->get_fd(), (EVENT_IN | EVENT_HUP | EVENT_NVAL), handler);
 192
 193         return true;
 194 }
 195
 196 bool channel::read_sync(message &msg)
 197 {
 198         message_header header;
 199         ssize_t size = 0;
 200         char buf[MAX_MSG_CAPACITY];
 201
 202         /* header */
 203         size = m_socket->recv(&header, sizeof(message_header), true);
 204         retv_if(size <= 0, false);
 205
 206         /* check error from header */
 207         if (m_handler && header.err != 0) {
 208                 m_handler->error_caught(this, header.err);
 209                 msg.header()->err = header.err;
 210                 return true;
 211         }
 212
 213         /* body */
 214         if (header.length > 0) {
 215                 size = m_socket->recv(&buf, header.length, true);
 216                 retv_if(size <= 0, false);
 217         }
 218
 219         buf[header.length] = '\0';
 220         msg.enclose(reinterpret_cast<const void *>(buf), header.length);
 221         msg.set_type(header.type);
 222         msg.header()->err = header.err;
 223
 224         if (m_handler)
 225                 m_handler->read(this, msg);
 226
 227         return true;
 228 }
 229
 230 bool channel::is_connected(void)
 231 {
 232         return m_connected.load();
 233 }
 234
 235 bool channel::set_option(int type, int value)
 236 {
 237         switch (type) {
 238         case SO_SNDBUF:
 239                 m_socket->set_buffer_size(type, value);
 240                 break;
 241         case SO_RCVBUF:
 242                 m_socket->set_buffer_size(type, value);
 243                 break;
 244         default:
 245                 break;
 246         }
 247
 248         return true;
 249 }
 250
 251 bool channel::get_option(int type, int &value) const
 252 {
 253         switch (type) {
 254         case 0:
 255                 value = m_socket->get_current_buffer_size();
 256                 break;
 257         case SO_SNDBUF:
 258                 value = m_socket->get_buffer_size(type);
 259                 break;
 260         case SO_RCVBUF:
 261                 value = m_socket->get_buffer_size(type);
 262                 break;
 263         default:
 264                 break;
 265         }
 266
 267         return true;
 268 }
 269
 270 int channel::get_fd(void) const
 271 {
 272         return m_fd;
 273 }
 274
 275 void channel::set_event_id(uint64_t id)
 276 {
 277         m_event_id = id;
 278 }