Improve subscription/unsubscription operation.
[apps/agl-service-can-low-level.git] / src / low-can-binding.cpp
1 /*
2  * Copyright (C) 2015, 2016 "IoT.bzh"
3  * Author "Romain Forlot" <romain.forlot@iot.bzh>
4  * Author "Loic Collignon" <loic.collignon@iot.bzh>
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 #include "low-can-binding.hpp"
20
21 #include <queue>
22 #include <vector>
23 #include <thread>
24 #include <fcntl.h>
25 #include <linux/can.h>
26 #include <json-c/json.h>
27 #include <systemd/sd-event.h>
28
29 #include "timer.hpp"
30 #include "openxc.pb.h"
31 #include "can-utils.hpp"
32 #include "can-signals.hpp"
33 #include "openxc-utils.hpp"
34
35 extern "C"
36 {
37         #include <afb/afb-binding.h>
38         #include <afb/afb-service-itf.h>
39 };
40
41 /*
42  *      Interface between the daemon and the binding
43  */
44 const struct afb_binding_interface *binder_interface;
45
46 /*
47  * CAN bus handler pointer. This is the object that will be use to
48  * initialize each CAN devices specified into the configuration file
49  *
50  * It is used by the reading thread also because of its can_message_q_ queue
51  * that store CAN messages read from the socket.
52  */
53 can_bus_t *can_bus_handler;
54
55 /********************************************************************************
56 *
57 *               Event management
58 *
59 *********************************************************************************/
60 int can_frame_received(sd_event_source *s, int fd, uint32_t revents, void *userdata)
61 {
62         can_bus_dev_t *can_bus_dev = (can_bus_dev_t*)userdata;
63
64         /* Notify reading thread that there is something to read */
65         if ((revents & EPOLLIN) != 0) {
66                 new_can_frame.notify_one();
67         }
68
69         /* check if error or hangup and reopen the socket and event_loop. 
70          * socket is protected by a mutex */
71         if ((revents & (EPOLLERR|EPOLLRDHUP|EPOLLHUP)) != 0)
72         {
73                 std::lock_guard<std::mutex> can_frame_lock(can_frame_mutex);
74                 sd_event_source_unref(s);
75                 can_bus_dev->close();
76                 can_bus_dev->open();
77                 can_bus_dev->start_reading(*can_bus_handler);
78                 can_bus_dev->event_loop_connection();
79         }
80
81         return 0;
82 }
83 /********************************************************************************
84 *
85 *               Subscription and unsubscription
86 *
87 *********************************************************************************/
88
89 static int make_subscription_unsubscription(struct afb_req request, std::map<std::string, struct afb_event>::iterator& ss_i, bool subscribe)
90 {
91         /* Make the subscription or unsubscription to the event */
92         if (((subscribe ? afb_req_subscribe : afb_req_unsubscribe)(request, ss_i->second)) < 0)
93         {
94                 ERROR(binder_interface, "Operation goes wrong for signal: %s", ss_i->first);
95                 return 0;
96         }
97         return 1;
98
99 }
100
101 static int create_event_handle(std::map<std::string, struct afb_event>::iterator& ss_i)
102 {
103         ss_i->second = afb_daemon_make_event(binder_interface->daemon, ss_i->first.c_str());
104         if (!afb_event_is_valid(ss_i->second)) 
105         {
106                 ERROR(binder_interface, "Can't create an event, something goes wrong.");
107                 return 0;
108         }
109
110         return 1;
111 }
112
113 static int subscribe_unsubscribe_signal(struct afb_req request, bool subscribe, const CanSignal& sig)
114 {
115         int ret;
116
117         // TODO: lock the subscribed_signals when insert/remove
118         auto ss_i = subscribed_signals.find(sig.genericName);
119         if (ss_i != subscribed_signals.end() && !afb_event_is_valid(ss_i->second))
120         {
121                 if(!subscribe)
122                 {
123                         NOTICE(binder_interface, "Event isn't valid, it can't be unsubscribed.");
124                         ret = -1;
125                 }
126                 else
127                 {
128                         /* Event it isn't valid annymore, recreate it */
129                         ret = create_event_handle(ss_i);
130                 }
131         }
132         else
133         {
134                 /* Event don't exist , so let's create it */
135                 struct afb_event empty_event = {nullptr, nullptr};
136                 auto ss_i = subscribed_signals.insert(std::make_pair(sig.genericName, empty_event));
137                 ret = create_event_handle(ss_i.first);
138         }
139
140         /* Check whether or not the event handler has been correctly created and
141          * make the subscription/unsubscription operation is so.
142          */
143         if (ret <= 0)
144                 return ret;
145         return make_subscription_unsubscription(request, ss_i, subscribe);
146 }
147
148 static int subscribe_unsubscribe_signals(struct afb_req request, bool subscribe, const std::vector<CanSignal>& signals)
149 {
150         int ret = 0;
151
152         for(const auto& signal_i : signals)
153         {
154                 ret = subscribe_unsubscribe_signal(request, subscribe, signal_i);
155                 if(ret == 0)
156                         return ret;
157         }
158         return ret;
159 }
160
161 static int subscribe_unsubscribe_all(struct afb_req request, bool subscribe)
162 {
163         int e = 0;
164
165         //for (const auto& sig : SIGNALS)
166         //      e += !subscribe_unsubscribe_signals(request, subscribe, sig);
167         e += !subscribe_unsubscribe_signals(request, subscribe, getSignals());
168         
169         return e == 0;
170 }
171
172 static int subscribe_unsubscribe_name(struct afb_req request, bool subscribe, const char *name)
173 {
174         std::vector<CanSignal> sig;
175         int ret = 0;
176
177         if (!::strcmp(name, "*"))
178                 ret = subscribe_unsubscribe_all(request, subscribe);
179         else
180         {
181                 //if(obd2_handler_c.is_obd2_signal(name))
182                 if(false)
183                 {
184                 // TODO
185                 }
186                 else
187                 {
188                         openxc_DynamicField search_key = build_DynamicField(name);
189                         sig = find_can_signals(search_key);
190                         if (sig.empty())
191                                 ret = 0;
192                 }
193                 ret = subscribe_unsubscribe_signals(request, subscribe, sig);
194         }
195         return ret;
196 }
197
198 static void subscribe_unsubscribe(struct afb_req request, bool subscribe)
199 {
200         int ok, i, n;
201         struct json_object *args, *a, *x;
202
203         /* makes the subscription/unsubscription */
204         args = afb_req_json(request);
205         if (args == NULL || !json_object_object_get_ex(args, "event", &a)) {
206                 ok = subscribe_unsubscribe_all(request, subscribe);
207         } else if (json_object_get_type(a) != json_type_array) {
208                 ok = subscribe_unsubscribe_name(request, subscribe, json_object_get_string(a));
209         } else {
210                 n = json_object_array_length(a);
211                 ok = 0;
212                 for (i = 0 ; i < n ; i++) {
213                         x = json_object_array_get_idx(a, i);
214                         if (subscribe_unsubscribe_name(request, subscribe, json_object_get_string(x)))
215                                 ok++;
216                 }
217                 ok = (ok == n);
218         }
219
220         /* send the report */
221         if (ok)
222                 afb_req_success(request, NULL, NULL);
223         else
224                 afb_req_fail(request, "error", NULL);
225 }
226
227 extern "C"
228 {
229         static void subscribe(struct afb_req request)
230         {
231                 subscribe_unsubscribe(request, true);
232         }
233
234         static void unsubscribe(struct afb_req request)
235         {
236                 subscribe_unsubscribe(request, false);
237         }
238
239         static const struct afb_verb_desc_v1 verbs[]=
240         {
241                 { .name= "subscribe",   .session= AFB_SESSION_NONE, .callback= subscribe,       .info= "subscribe to notification of CAN bus messages." },
242                 { .name= "unsubscribe", .session= AFB_SESSION_NONE, .callback= unsubscribe,     .info= "unsubscribe a previous subscription." }
243         };
244
245         static const struct afb_binding binding_desc {
246                 AFB_BINDING_VERSION_1,
247                 {
248                         "CAN bus service",
249                         "can",
250                         verbs
251                 }
252         };
253
254         const struct afb_binding *afbBindingV1Register (const struct afb_binding_interface *itf)
255         {
256                 binder_interface = itf;
257
258                 return &binding_desc;
259         }
260
261         /**
262         * @brief Initialize the binding.
263         * 
264         * @param[in] service Structure which represent the Application Framework Binder.
265         * 
266         * @return Exit code, zero if success.
267         */
268         int afbBindingV1ServiceInit(struct afb_service service)
269         {
270                 int fd_conf;
271                 fd_conf = afb_daemon_rootdir_open_locale(binder_interface->daemon, "can_bus.json", O_RDONLY, NULL);
272
273                 /* Initialize the CAN bus handler */
274                 can_bus_t cbh(fd_conf);
275                 can_bus_handler = &cbh;
276
277                 /* Open CAN socket */
278                 if(can_bus_handler->init_can_dev() == 0)
279                 {
280                         can_bus_handler->start_threads();
281                         return 0;
282                 }
283                 ERROR(binder_interface, "There was something wrong with CAN device Initialization. Check your config file maybe");
284                 return 1;
285         }
286 };