plugins/murphyplugin/murphysource.cpp

   1 /*
   2     Copyright (C) 2012  Intel Corporation
   3
   4     This library is free software; you can redistribute it and/or
   5     modify it under the terms of the GNU Lesser General Public
   6     License as published by the Free Software Foundation; either
   7     version 2.1 of the License, or (at your option) any later version.
   8
   9     This library is distributed in the hope that it will be useful,
  10     but WITHOUT ANY WARRANTY; without even the implied warranty of
  11     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12     Lesser General Public License for more details.
  13
  14     You should have received a copy of the GNU Lesser General Public
  15     License along with this library; if not, write to the Free Software
  16     Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
  17 */
  18
  19 #include "murphysource.h"
  20
  21 #include <iostream>
  22 #include <boost/assert.hpp>
  23 #include <boost/lexical_cast.hpp>
  24 #include <glib.h>
  25 #include <sstream>
  26 #include <listplusplus.h>
  27 #include <timestamp.h>
  28
  29 #include <murphy/common.h>
  30
  31 /*#ifdef USE_QT_CORE
  32 #include <murphy/qt/qt-glue.h>
  33 #else*/
  34
  35 #include <murphy/glib/glib-glue.h>
  36 //#endif
  37
  38 // #include <vehicleproperty.h>
  39 // #include <abstractpropertytype.h>
  40
  41 #include "debugout.h"
  42
  43
  44 void MurphySource::processValue(string propertyName, AbstractPropertyType *prop)
  45 {
  46     if (murphyProperties.find(propertyName) != murphyProperties.end()) {
  47         delete murphyProperties[propertyName];
  48         murphyProperties[propertyName] = prop;
  49     }
  50     else {
  51         murphyProperties[propertyName] = prop;
  52                 m_re->updateSupported(supported(), PropertyList(), this);
  53     }
  54
  55     // cout << "updating property!" << endl;
  56     m_re->updateProperty(propertyName, prop, uuid());
  57 }
  58
  59 bool MurphySource::hasProperty(string propertyName)
  60 {
  61     PropertyList props = supported();
  62     return ListPlusPlus<VehicleProperty::Property>(&props).contains(propertyName);
  63 }
  64
  65 static void recvfrom_msg(mrp_transport_t *transp, mrp_msg_t *msg,
  66                          mrp_sockaddr_t *addr, socklen_t addrlen,
  67                          void *user_data)
  68 {
  69     MurphySource *s = (MurphySource *) user_data;
  70
  71     void *cursor = NULL;
  72     uint16_t tag = 0;
  73     uint16_t type = 0;
  74     mrp_msg_value_t value;
  75     size_t size;
  76
  77     char *property_name;
  78
  79         DebugOut()<<"Received a message from Murphy!"<<endl;
  80
  81     if (!mrp_msg_iterate(msg, &cursor, &tag, &type, &value, &size))
  82         return;
  83
  84     if (tag == 1 && type == MRP_MSG_FIELD_STRING)
  85         property_name = value.str;
  86
  87     string dstr(property_name);
  88
  89     if (!mrp_msg_iterate(msg, &cursor, &tag, &type, &value, &size))
  90         return;
  91
  92     if (tag != 2)
  93         return;
  94
  95         DebugOut() << "Property '" << property_name << "' with value: " <<endl;
  96
  97     switch (type) {
  98         case MRP_MSG_FIELD_STRING:
  99         {
 100                         StringPropertyType *prop = new StringPropertyType(dstr,value.str);
 101
 102             if (!s->hasProperty(dstr)) {
 103                 VehicleProperty::registerProperty(dstr,
 104                                                 [dstr](){return new StringPropertyType(dstr,"");});
 105             }
 106
 107                         DebugOut() << "string:" << value.str << std::endl;
 108             s->processValue(property_name, prop);
 109             break;
 110         }
 111         case MRP_MSG_FIELD_DOUBLE:
 112         {
 113             BasicPropertyType<double> *prop =
 114                                         new BasicPropertyType<double>(dstr,value.dbl);
 115
 116             if (!s->hasProperty(dstr)) {
 117                 VehicleProperty::registerProperty(dstr,
 118                                                 [dstr](){return new BasicPropertyType<double>(dstr,0);});
 119             }
 120
 121                         DebugOut() << "double:" << value.dbl << std::endl;
 122             s->processValue(property_name, prop);
 123             break;
 124         }
 125         case MRP_MSG_FIELD_BOOL:
 126         {
 127             BasicPropertyType<bool> *prop =
 128                                         new BasicPropertyType<bool>(dstr,value.bln);
 129
 130             if (!s->hasProperty(dstr)) {
 131                 VehicleProperty::registerProperty(dstr,
 132                                                 [dstr](){return new BasicPropertyType<bool>(dstr,FALSE);});
 133             }
 134
 135                         DebugOut() << "boolean:" << value.bln << std::endl;
 136             s->processValue(property_name, prop);
 137             break;
 138         }
 139         case MRP_MSG_FIELD_UINT32:
 140         {
 141             BasicPropertyType<uint32_t> *prop =
 142                                         new BasicPropertyType<uint32_t>(dstr,value.u32);
 143
 144             if (!s->hasProperty(dstr)) {
 145                 VehicleProperty::registerProperty(dstr,
 146                                                 [dstr](){return new BasicPropertyType<uint32_t>(dstr,0);});
 147             }
 148
 149                         DebugOut() << "uint32:" << value.u32 << std::endl;
 150             s->processValue(property_name, prop);
 151             break;
 152         }
 153         case MRP_MSG_FIELD_UINT16:
 154         {
 155             BasicPropertyType<uint16_t> *prop =
 156                                         new BasicPropertyType<uint16_t>(dstr,value.u16);
 157
 158             if (!s->hasProperty(dstr)) {
 159                 VehicleProperty::registerProperty(dstr,
 160                                                 [dstr](){return new BasicPropertyType<uint16_t>(dstr,0);});
 161             }
 162
 163                         DebugOut() << "uint16:" << value.u16 << std::endl;
 164             s->processValue(property_name, prop);
 165             break;
 166         }
 167         case MRP_MSG_FIELD_INT32:
 168         {
 169             BasicPropertyType<int32_t> *prop =
 170                                         new BasicPropertyType<int32_t>(dstr, value.s32);
 171
 172             if (!s->hasProperty(dstr)) {
 173                 VehicleProperty::registerProperty(dstr,
 174                                                 [dstr](){return new BasicPropertyType<int32_t>(dstr,0);});
 175             }
 176
 177                         DebugOut() << "int32:" << value.s32 << std::endl;
 178             s->processValue(property_name, prop);
 179             break;
 180         }
 181         case MRP_MSG_FIELD_INT16:
 182         {
 183             BasicPropertyType<int16_t> *prop =
 184                                         new BasicPropertyType<int16_t>(dstr, value.s16);
 185
 186             if (!s->hasProperty(dstr)) {
 187                 VehicleProperty::registerProperty(dstr,
 188                                                 [dstr](){return new BasicPropertyType<int16_t>(dstr, 0);});
 189             }
 190
 191                         DebugOut() << "int16:" << value.s16 << std::endl;
 192             s->processValue(property_name, prop);
 193             break;
 194         }
 195 #if 0
 196         case MRP_MSG_FIELD_UINT8:
 197         {
 198             BasicPropertyType<uint8_t> prop =
 199                     BasicPropertyType<uint8_t>(value.u8);
 200             std::cout << "byte:" << value.u8 << std::endl;
 201             s->processValue(property_name, &prop);
 202             break;
 203         }
 204 #endif
 205         default:
 206                         DebugOut()<<"Unknown type"<<endl;
 207     }
 208 }
 209
 210 static void recv_msg(mrp_transport_t *transp, mrp_msg_t *msg, void *user_data)
 211 {
 212     return recvfrom_msg(transp, msg, NULL, 0, user_data);
 213 }
 214
 215 static void closed_evt(mrp_transport_t *t, int error, void *user_data)
 216 {
 217    /* TODO: should process the error somehow */
 218
 219     MurphySource *s = (MurphySource *) user_data;
 220
 221     s->setState(MRP_PROCESS_STATE_NOT_READY);
 222     s->setConnected(FALSE);
 223 }
 224
 225
 226 void MurphySource::setConnected(bool connected)
 227 {
 228     m_connected = connected;
 229 }
 230
 231
 232 int MurphySource::connectToMurphy()
 233 {
 234     mrp_sockaddr_t addr;
 235     socklen_t alen;
 236     int flags = MRP_TRANSPORT_REUSEADDR | MRP_TRANSPORT_MODE_MSG;
 237     static mrp_transport_evt_t evt;
 238     const char *atype;
 239
 240     evt.recvmsg = recv_msg;
 241     evt.recvmsgfrom = recvfrom_msg;
 242     evt.closed = closed_evt;
 243
 244     if (m_ml == NULL || m_address.empty())
 245         return -1;
 246
 247     if (m_connected == TRUE)
 248         return -1;
 249
 250     alen = mrp_transport_resolve(NULL, m_address.c_str(), &addr, sizeof(addr), &atype);
 251
 252     if (alen <= 0) {
 253         debugOut("Failed to resolve address");
 254         return -1;
 255     }
 256
 257     m_tport = mrp_transport_create(m_ml, atype, &evt, this, flags);
 258
 259     if (!m_tport) {
 260         debugOut("Can't create a Murphy transport");
 261         return -1;
 262     }
 263
 264     if (mrp_transport_connect(m_tport, &addr, alen) == 0) {
 265         mrp_transport_destroy(m_tport);
 266         m_tport = NULL;
 267         debugOut("Failed to connect to Murphy");
 268         return -1;
 269     }
 270
 271     setConnected(true);
 272
 273     return 0;
 274 }
 275
 276
 277 MurphySource::MurphySource(AbstractRoutingEngine *re, map<string, string> config) : AbstractSource(re, config)
 278 {
 279     m_source = this;
 280     m_re = re;
 281     m_connected = false;
 282
 283     // main loop integration
 284
 285 /*#ifdef USE_QT_CORE
 286     m_ml = mrp_mainloop_qt_get();
 287     debugOut("Murphy plugin initialized using QT mainloop!");
 288 #else*/
 289     GMainLoop *g_ml = g_main_loop_new(NULL, TRUE);
 290     m_ml = mrp_mainloop_glib_get(g_ml);
 291     debugOut("Murphy plugin initialized using glib mainloop!");
 292 //#endif
 293
 294     setConfiguration(config);
 295 }
 296
 297 MurphySource::~MurphySource()
 298 {
 299     mrp_process_set_state("ambd", MRP_PROCESS_STATE_NOT_READY);
 300     mrp_transport_destroy(m_tport);
 301     mrp_mainloop_unregister(m_ml);
 302
 303     map<string, AbstractPropertyType *>::iterator i;
 304
 305     for (i = murphyProperties.begin(); i != murphyProperties.end(); i++) {
 306         // TODO: unregister VehicleProperty (*i).first
 307
 308         delete (*i).second;
 309     }
 310 }
 311
 312 void MurphySource::setState(mrp_process_state_t state)
 313 {
 314     m_state = state;
 315 }
 316
 317 mrp_process_state_t MurphySource::getState()
 318 {
 319     return m_state;
 320 }
 321
 322 static void murphy_watch(const char *id, mrp_process_state_t state, void *user_data)
 323 {
 324     MurphySource *s = (MurphySource *) user_data;
 325
 326     debugOut("murphy process watch event");
 327
 328     if (strcmp(id, "murphy-amb") != 0)
 329         return;
 330
 331     printf("murphyd state changed to %s\n",
 332             state == MRP_PROCESS_STATE_READY ? "ready" : "not ready");
 333
 334     if (state == MRP_PROCESS_STATE_NOT_READY &&
 335             s->getState() == MRP_PROCESS_STATE_READY) {
 336                 DebugOut()<<"lost connection to murphyd"<<endl;
 337     }
 338
 339     else if (state == MRP_PROCESS_STATE_READY) {
 340         /* start connecting if needed */
 341         s->connectToMurphy();
 342     }
 343
 344     s->setState(state);
 345 }
 346
 347 void MurphySource::readyToConnect(mrp_mainloop_t *ml)
 348 {
 349     /* set a watch to follow Murphy status */
 350
 351     if (mrp_process_set_watch("murphy-amb", ml, murphy_watch, this) < 0) {
 352                 DebugOut()<<"failed to set a murphy process watch"<<endl;
 353         return;
 354     }
 355
 356     mrp_process_set_state("ambd", MRP_PROCESS_STATE_READY);
 357
 358     /* check if Murphy is running */
 359     m_state = mrp_process_query_state("murphy-amb");
 360
 361     if (m_state == MRP_PROCESS_STATE_READY)
 362         connectToMurphy();
 363 }
 364
 365
 366 void MurphySource::setConfiguration(map<string, string> config)
 367 {
 368     string address;
 369
 370     for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++) {
 371         if ((*i).first == "address") {
 372             address = (*i).second;
 373             // cout << "address: " << address << endl;
 374
 375             // TODO: sanity check
 376             m_address = address;
 377         }
 378     }
 379
 380     // TODO: read supported values from configuration?
 381         m_re->updateSupported(supported(), PropertyList(), this);
 382
 383     // set up the connection with Murphy
 384     if (!m_address.empty())
 385         readyToConnect(m_ml);
 386 }
 387
 388
 389 PropertyList MurphySource::supported()
 390 {
 391     // debugOut("> supported");
 392
 393     PropertyList properties;
 394     map<string, AbstractPropertyType *>::iterator i;
 395
 396     for (i = murphyProperties.begin(); i != murphyProperties.end(); i++) {
 397         properties.push_back((*i).first);
 398     }
 399
 400     return properties;
 401 }
 402
 403
 404 int MurphySource::supportedOperations()
 405 {
 406     // debugOut("> supportedOperations");
 407         return Get; // | Set;
 408         //we really don't support Set
 409 }
 410
 411
 412 void MurphySource::subscribeToPropertyChanges(VehicleProperty::Property property)
 413 {
 414     // debugOut("> subscribeToPropertyChanges");
 415 }
 416
 417
 418 void MurphySource::unsubscribeToPropertyChanges(VehicleProperty::Property property)
 419 {
 420     // debugOut("> unsubscribeToPropertyChanges");
 421 }
 422
 423
 424 void MurphySource::getPropertyAsync(AsyncPropertyReply *reply)
 425 {
 426     // debugOut("> getPropertyAsync");
 427
 428     if (murphyProperties.find(reply->property) != murphyProperties.end()) {
 429         AbstractPropertyType *prop = murphyProperties[reply->property];
 430         reply->value = prop;
 431                 reply->success = true;
 432         reply->completed(reply);
 433     }
 434 }
 435
 436
 437 void MurphySource::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 438 {
 439     // debugOut("> getRangePropertyAsync");
 440 }
 441
 442
 443 AsyncPropertyReply *MurphySource::setProperty(AsyncSetPropertyRequest request)
 444 {
 445     // debugOut("> setProperty");
 446
 447     processValue(request.property, request.value);
 448
 449     AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 450     reply->success = true;
 451     reply->completed(reply);
 452     return reply;
 453 }
 454
 455 #if 1
 456 extern "C" AbstractSource *create(AbstractRoutingEngine* routingengine, map<string, string> config)
 457 {
 458     return new MurphySource(routingengine, config);
 459 }
 460 #endif