src/server/sensor_event_dispatcher.cpp

   1 /*
   2  * libsensord-share
   3  *
   4  * Copyright (c) 2014 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 <command_common.h>
  21 #include <sensor_event_dispatcher.h>
  22 #include <sensor_logs.h>
  23 #include <thread>
  24
  25 using std::thread;
  26 using std::vector;
  27 using std::pair;
  28
  29 #define MAX_PENDING_CONNECTION 32
  30
  31 sensor_event_dispatcher::sensor_event_dispatcher()
  32 : m_lcd_on(false)
  33 {
  34 }
  35
  36 sensor_event_dispatcher::~sensor_event_dispatcher() { }
  37
  38
  39 sensor_event_dispatcher& sensor_event_dispatcher::get_instance()
  40 {
  41         static sensor_event_dispatcher inst;
  42         return inst;
  43 }
  44
  45
  46 bool sensor_event_dispatcher::run(void)
  47 {
  48         INFO("Starting Event Dispatcher\n");
  49
  50         if (!m_accept_socket.create(SOCK_SEQPACKET)) {
  51                 ERR("Listener Socket Creation failed in Server \n");
  52                 return false;
  53         }
  54
  55         if(!m_accept_socket.bind(EVENT_CHANNEL_PATH)) {
  56                 ERR("Listener Socket Binding failed in Server \n");
  57                 m_accept_socket.close();
  58                 return false;
  59         }
  60
  61         if(!m_accept_socket.listen(MAX_PENDING_CONNECTION)) {
  62                 ERR("Socket Listen failed in Server \n");
  63                 return false;
  64         }
  65
  66         thread accepter(&sensor_event_dispatcher::accept_connections, this);
  67         accepter.detach();
  68
  69         thread dispatcher(&sensor_event_dispatcher::dispatch_event, this);
  70         dispatcher.detach();
  71
  72         return true;
  73 }
  74
  75 void sensor_event_dispatcher::accept_event_channel(csocket client_socket)
  76 {
  77         int client_id;
  78         event_channel_ready_t event_channel_ready;
  79         client_info_manager& client_info_manager = get_client_info_manager();
  80
  81         client_socket.set_connection_mode();
  82
  83         if (client_socket.recv(&client_id, sizeof(client_id)) <= 0) {
  84                 ERR("Failed to receive client id on socket fd[%d]", client_socket.get_socket_fd());
  85                 return;
  86         }
  87
  88         client_socket.set_transfer_mode();
  89
  90         AUTOLOCK(m_mutex);
  91
  92         if(!get_client_info_manager().set_event_socket(client_id, client_socket)) {
  93                 ERR("Failed to store event socket[%d] for %s", client_socket.get_socket_fd(),
  94                         client_info_manager.get_client_info(client_id));
  95                 return;
  96         }
  97
  98         event_channel_ready.magic = EVENT_CHANNEL_MAGIC;
  99         event_channel_ready.client_id = client_id;
 100
 101         INFO("Event channel is accepted for %s on socket[%d]",
 102                 client_info_manager.get_client_info(client_id), client_socket.get_socket_fd());
 103
 104         if (client_socket.send(&event_channel_ready, sizeof(event_channel_ready)) <= 0) {
 105                 ERR("Failed to send event_channel_ready packet to %s on socket fd[%d]",
 106                         client_info_manager.get_client_info(client_id), client_socket.get_socket_fd());
 107                 return;
 108         }
 109 }
 110
 111 void sensor_event_dispatcher::accept_connections(void)
 112 {
 113         INFO("Event channel acceptor is started.\n");
 114
 115         while (true) {
 116                 csocket client_socket;
 117
 118                 if (!m_accept_socket.accept(client_socket)) {
 119                         ERR("Accepting socket failed in Server \n");
 120                         continue;
 121                 }
 122
 123                 INFO("New client connected (socket_fd : %d)\n", client_socket.get_socket_fd());
 124
 125                 thread event_channel_creator(&sensor_event_dispatcher::accept_event_channel, this, client_socket);
 126                 event_channel_creator.detach();
 127         }
 128 }
 129
 130 void sensor_event_dispatcher::dispatch_event(void)
 131 {
 132         const int MAX_SYNTH_PER_SENSOR = 5;
 133
 134         vector<sensor_event_t> v_sensor_events(MAX_SYNTH_PER_SENSOR);
 135
 136         INFO("Event Dispatcher started");
 137
 138         while (true) {
 139                 void *seed_event = get_event_queue().pop();
 140
 141                 vector<void *> sensor_events;
 142                 sensor_events.push_back(seed_event);
 143
 144                 virtual_sensors v_sensors = get_active_virtual_sensors();
 145
 146                 auto it_v_sensor = v_sensors.begin();
 147
 148                 while (it_v_sensor != v_sensors.end()) {
 149                         int synthesized_cnt;
 150                         v_sensor_events.clear();
 151                         (*it_v_sensor)->synthesize(*((sensor_event_t *)seed_event));
 152                         synthesized_cnt = v_sensor_events.size();
 153
 154                         for (int i = 0; i < synthesized_cnt; ++i) {
 155                                 sensor_event_t *v_event = (sensor_event_t*)malloc(sizeof(sensor_event_t));
 156                                 if (!v_event) {
 157                                         ERR("Failed to allocate memory");
 158                                         continue;
 159                                 }
 160
 161                                 memcpy(v_event, &v_sensor_events[i], sizeof(sensor_event_t));
 162                                 sensor_events.push_back(v_event);
 163                         }
 164
 165                         ++it_v_sensor;
 166                 }
 167
 168                 sort_sensor_events(sensor_events);
 169
 170                 for (unsigned int i = 0; i < sensor_events.size(); ++i) {
 171                         if (is_record_event(((sensor_event_t *)(sensor_events[i]))->event_type))
 172                                 put_last_event(((sensor_event_t *)(sensor_events[i]))->event_type, *((sensor_event_t *)(sensor_events[i])));
 173                 }
 174
 175                 send_sensor_events(sensor_events);
 176         }
 177 }
 178
 179
 180 void sensor_event_dispatcher::send_sensor_events(vector<void *> &events)
 181 {
 182         void *event;
 183         sensor_event_t *sensor_events = NULL;
 184         client_info_manager& client_info_manager = get_client_info_manager();
 185
 186         const int RESERVED_CLIENT_CNT = 20;
 187         static client_id_vec id_vec(RESERVED_CLIENT_CNT);
 188
 189         for (unsigned int i = 0; i < events.size(); ++i) {
 190                 sensor_id_t sensor_id;
 191                 unsigned int event_type;
 192                 int length;
 193
 194                 sensor_events = (sensor_event_t*)events[i];
 195                 length = sizeof(sensor_event_t) + sensor_events->data_length;
 196                 sensor_id = sensor_events->sensor_id;
 197                 event_type = sensor_events->event_type;
 198
 199                 event = (void *)malloc(length);
 200                 if (!event) {
 201                         ERR("Failed to allocate memory");
 202                         return;
 203                 }
 204
 205                 memcpy(event, sensor_events, sizeof(sensor_event_t));
 206                 memcpy((char *)event + sizeof(sensor_event_t), sensor_events->data, sensor_events->data_length);
 207
 208                 id_vec.clear();
 209                 client_info_manager.get_listener_ids(sensor_id, event_type, id_vec);
 210
 211                 auto it_client_id = id_vec.begin();
 212
 213                 while (it_client_id != id_vec.end()) {
 214                         csocket client_socket;
 215                         client_info_manager.get_event_socket(*it_client_id, client_socket);
 216                         bool ret = (client_socket.send(event, length) > 0);
 217
 218                         if (ret)
 219                                 DBG("Event[0x%x] sent to %s on socket[%d]", event_type, client_info_manager.get_client_info(*it_client_id), client_socket.get_socket_fd());
 220                         else
 221                                 ERR("Failed to send event[0x%x] to %s on socket[%d]", event_type, client_info_manager.get_client_info(*it_client_id), client_socket.get_socket_fd());
 222
 223                         ++it_client_id;
 224                 }
 225
 226                 free(sensor_events->data);
 227                 free(sensor_events);
 228         }
 229 }
 230
 231 client_info_manager& sensor_event_dispatcher::get_client_info_manager(void)
 232 {
 233         return client_info_manager::get_instance();
 234 }
 235
 236 sensor_event_queue& sensor_event_dispatcher::get_event_queue(void)
 237 {
 238         return sensor_event_queue::get_instance();
 239 }
 240
 241 bool sensor_event_dispatcher::is_record_event(unsigned int event_type)
 242 {
 243         return false;
 244 }
 245
 246 void sensor_event_dispatcher::put_last_event(unsigned int event_type, const sensor_event_t &event)
 247 {
 248         AUTOLOCK(m_last_events_mutex);
 249         m_last_events[event_type] = event;
 250 }
 251
 252 bool sensor_event_dispatcher::get_last_event(unsigned int event_type, sensor_event_t &event)
 253 {
 254         AUTOLOCK(m_last_events_mutex);
 255
 256         auto it_event = m_last_events.find(event_type);
 257
 258         if (it_event == m_last_events.end())
 259                 return false;
 260
 261         event = it_event->second;
 262         return true;
 263 }
 264
 265 bool sensor_event_dispatcher::has_active_virtual_sensor(virtual_sensor *sensor)
 266 {
 267         AUTOLOCK(m_active_virtual_sensors_mutex);
 268
 269         auto it_v_sensor = find(m_active_virtual_sensors.begin(), m_active_virtual_sensors.end(), sensor);
 270
 271         return (it_v_sensor != m_active_virtual_sensors.end());
 272 }
 273
 274
 275 virtual_sensors sensor_event_dispatcher::get_active_virtual_sensors(void)
 276 {
 277         AUTOLOCK(m_active_virtual_sensors_mutex);
 278
 279         return m_active_virtual_sensors;
 280 }
 281
 282 struct sort_comp {
 283         bool operator()(const void *left, const void *right) {
 284                 return ((sensor_event_t *)left)->data->timestamp < ((sensor_event_t *)right)->data->timestamp;
 285         }
 286 };
 287
 288 void sensor_event_dispatcher::sort_sensor_events(vector<void *> &events)
 289 {
 290         std::sort(events.begin(), events.end(), sort_comp());
 291 }
 292
 293 void sensor_event_dispatcher::request_last_event(int client_id, sensor_id_t sensor_id)
 294 {
 295         client_info_manager& client_info_manager = get_client_info_manager();
 296         event_type_vector event_vec;
 297         csocket client_socket;
 298
 299         if (client_info_manager.get_registered_events(client_id, sensor_id, event_vec)) {
 300                 if (!client_info_manager.get_event_socket(client_id, client_socket)) {
 301                         ERR("Failed to get event socket from %s",
 302                                         client_info_manager.get_client_info(client_id));
 303                         return;
 304                 }
 305
 306                 auto it_event = event_vec.begin();
 307                 while (it_event != event_vec.end()) {
 308                         sensor_event_t event;
 309                         if (is_record_event(*it_event) && get_last_event(*it_event, event)) {
 310                                 if (client_socket.send(&event, sizeof(event)) > 0)
 311                                         INFO("Send the last event[0x%x] to %s on socket[%d]", event.event_type,
 312                                                 client_info_manager.get_client_info(client_id), client_socket.get_socket_fd());
 313                                 else
 314                                         ERR("Failed to send event[0x%x] to %s on socket[%d]", event.event_type,
 315                                                 client_info_manager.get_client_info(client_id), client_socket.get_socket_fd());
 316                         }
 317                         ++it_event;
 318                 }
 319         }
 320 }
 321
 322
 323 bool sensor_event_dispatcher::add_active_virtual_sensor(virtual_sensor * sensor)
 324 {
 325         AUTOLOCK(m_active_virtual_sensors_mutex);
 326
 327         if (has_active_virtual_sensor(sensor)) {
 328                 ERR("[%s] sensor is already added on active virtual sensors", sensor->get_name());
 329                 return false;
 330         }
 331
 332         m_active_virtual_sensors.push_back(sensor);
 333
 334         return true;
 335 }
 336
 337 bool sensor_event_dispatcher::delete_active_virtual_sensor(virtual_sensor * sensor)
 338 {
 339         AUTOLOCK(m_active_virtual_sensors_mutex);
 340
 341         auto it_v_sensor = find(m_active_virtual_sensors.begin(), m_active_virtual_sensors.end(), sensor);
 342
 343         if (it_v_sensor == m_active_virtual_sensors.end()) {
 344                 ERR("Fail to delete non-existent [%s] sensor on active virtual sensors", sensor->get_name());
 345                 return false;
 346         }
 347
 348         m_active_virtual_sensors.erase(it_v_sensor);
 349
 350         return true;
 351 }