src/server/sensor_event_dispatcher.cpp

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