cpp/src/IceStorm/TransientTopicI.cpp

   1 // **********************************************************************
   2 //
   3 // Copyright (c) 2003-2011 ZeroC, Inc. All rights reserved.
   4 //
   5 // This copy of Ice is licensed to you under the terms described in the
   6 // ICE_LICENSE file included in this distribution.
   7 //
   8 // **********************************************************************
   9
  10 #include <IceUtil/DisableWarnings.h>
  11 #include <IceStorm/TransientTopicI.h>
  12 #include <IceStorm/Instance.h>
  13 #include <IceStorm/Subscriber.h>
  14 #include <IceStorm/TraceLevels.h>
  15 #include <IceStorm/Util.h>
  16
  17 #include <Ice/Ice.h>
  18
  19 #include <list>
  20 #include <algorithm>
  21
  22 using namespace IceStorm;
  23 using namespace std;
  24
  25 namespace
  26 {
  27
  28 //
  29 // The servant has a 1-1 association with a topic. It is used to
  30 // receive events from Publishers.
  31 //
  32 class TransientPublisherI : public Ice::BlobjectArray
  33 {
  34 public:
  35
  36     TransientPublisherI(const TransientTopicImplPtr& impl) :
  37         _impl(impl)
  38     {
  39     }
  40
  41     virtual bool
  42     ice_invoke(const pair<const Ice::Byte*, const Ice::Byte*>& inParams,
  43                Ice::ByteSeq&,
  44                const Ice::Current& current)
  45     {
  46         // Use cached reads.
  47         EventDataPtr event = new EventData(
  48             current.operation,
  49             current.mode,
  50             Ice::ByteSeq(),
  51             current.ctx);
  52
  53         //
  54         // COMPILERBUG: gcc 4.0.1 doesn't like this.
  55         //
  56         //event->data.swap(Ice::ByteSeq(inParams.first, inParams.second));
  57         Ice::ByteSeq data(inParams.first, inParams.second);
  58         event->data.swap(data);
  59
  60         EventDataSeq v;
  61         v.push_back(event);
  62         _impl->publish(false, v);
  63
  64         return true;
  65     }
  66
  67 private:
  68
  69     const TransientTopicImplPtr _impl;
  70 };
  71
  72 //
  73 // The servant has a 1-1 association with a topic. It is used to
  74 // receive events from linked Topics.
  75 //
  76 class TransientTopicLinkI : public TopicLink
  77 {
  78 public:
  79
  80     TransientTopicLinkI(const TransientTopicImplPtr& impl) :
  81         _impl(impl)
  82     {
  83     }
  84
  85     virtual void
  86     forward(const EventDataSeq& v, const Ice::Current& current)
  87     {
  88         _impl->publish(true, v);
  89     }
  90
  91 private:
  92
  93     const TransientTopicImplPtr _impl;
  94 };
  95
  96 }
  97
  98 namespace IceStorm
  99 {
 100 extern string identityToTopicName(const Ice::Identity& id);
 101 }
 102
 103 TransientTopicImpl::TransientTopicImpl(
 104     const InstancePtr& instance,
 105     const string& name,
 106     const Ice::Identity& id) :
 107     _instance(instance),
 108     _name(name),
 109     _id(id),
 110     _destroyed(false)
 111 {
 112     //
 113     // Create a servant per topic to receive event data. If the
 114     // category is empty then we are in backwards compatibility
 115     // mode. In this case the servant's identity is
 116     // category=<topicname>, name=publish, otherwise the name is
 117     // <instancename>/<topicname>.publish. The same applies to the
 118     // link proxy.
 119     //
 120     // Activate the object and save a reference to give to publishers.
 121     //
 122     Ice::Identity pubid;
 123     Ice::Identity linkid;
 124     if(id.category.empty())
 125     {
 126         pubid.category = _name;
 127         pubid.name = "publish";
 128         linkid.category = _name;
 129         linkid.name = "link";
 130     }
 131     else
 132     {
 133         pubid.category = id.category;
 134         pubid.name = _name + ".publish";
 135         linkid.category = id.category;
 136         linkid.name = _name + ".link";
 137     }
 138
 139     _publisherPrx = _instance->publishAdapter()->add(new TransientPublisherI(this), pubid);
 140     _linkPrx = TopicLinkPrx::uncheckedCast(_instance->publishAdapter()->add(new TransientTopicLinkI(this), linkid));
 141 }
 142
 143 TransientTopicImpl::~TransientTopicImpl()
 144 {
 145 }
 146
 147 string
 148 TransientTopicImpl::getName(const Ice::Current&) const
 149 {
 150     // Immutable
 151     return _name;
 152 }
 153
 154 Ice::ObjectPrx
 155 TransientTopicImpl::getPublisher(const Ice::Current&) const
 156 {
 157     // Immutable
 158     return _publisherPrx;
 159 }
 160
 161 Ice::ObjectPrx
 162 TransientTopicImpl::getNonReplicatedPublisher(const Ice::Current&) const
 163 {
 164     // Immutable
 165     return _publisherPrx;
 166 }
 167
 168 //
 169 // COMPILERFIX: For some reason with VC6 find reports an error.
 170 //
 171 #if defined(_MSC_VER) && (_MSC_VER < 1300)
 172 namespace
 173 {
 174 static vector<SubscriberPtr>::iterator
 175 find(vector<SubscriberPtr>::iterator start, vector<SubscriberPtr>::iterator end, const Ice::Identity& ident)
 176 {
 177     while(start != end)
 178     {
 179         if(*start == ident)
 180         {
 181             return start;
 182         }
 183         ++start;
 184     }
 185     return end;
 186 }
 187 }
 188 #endif
 189
 190 void
 191 TransientTopicImpl::subscribe(const QoS& origQoS, const Ice::ObjectPrx& obj, const Ice::Current&)
 192 {
 193     Ice::Identity id = obj->ice_getIdentity();
 194     TraceLevelsPtr traceLevels = _instance->traceLevels();
 195     QoS qos = origQoS;
 196     if(traceLevels->topic > 0)
 197     {
 198         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 199         out << _name << ": subscribe: " << _instance->communicator()->identityToString(id);
 200
 201         if(traceLevels->topic > 1)
 202         {
 203             out << " endpoints: " << IceStormInternal::describeEndpoints(obj)
 204                 << " QoS: ";
 205             for(QoS::const_iterator p = qos.begin(); p != qos.end() ; ++p)
 206             {
 207                 if(p != qos.begin())
 208                 {
 209                     out << ',';
 210                 }
 211                 out << '[' << p->first << "," << p->second << ']';
 212             }
 213         }
 214     }
 215
 216     string reliability = "oneway";
 217     {
 218         QoS::iterator p = qos.find("reliability");
 219         if(p != qos.end())
 220         {
 221             reliability = p->second;
 222             qos.erase(p);
 223         }
 224     }
 225
 226     Ice::ObjectPrx newObj = obj;
 227     if(reliability == "batch")
 228     {
 229         if(newObj->ice_isDatagram())
 230         {
 231             newObj = newObj->ice_batchDatagram();
 232         }
 233         else
 234         {
 235             newObj = newObj->ice_batchOneway();
 236         }
 237     }
 238     else if(reliability == "twoway")
 239     {
 240         newObj = newObj->ice_twoway();
 241     }
 242     else if(reliability == "twoway ordered")
 243     {
 244         qos["reliability"] = "ordered";
 245         newObj = newObj->ice_twoway();
 246     }
 247     else // reliability == "oneway"
 248     {
 249         if(reliability != "oneway" && traceLevels->subscriber > 0)
 250         {
 251             Ice::Trace out(traceLevels->logger, traceLevels->subscriberCat);
 252             out << reliability <<" mode not understood.";
 253         }
 254         if(!newObj->ice_isDatagram())
 255         {
 256             newObj = newObj->ice_oneway();
 257         }
 258     }
 259
 260     Lock sync(*this);
 261     SubscriberRecord record;
 262     record.id = id;
 263     record.obj = newObj;
 264     record.theQoS = qos;
 265     record.topicName = _name;
 266     record.link = false;
 267     record.cost = 0;
 268
 269     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 270     if(p != _subscribers.end())
 271     {
 272         // If we already have this subscriber remove it from our
 273         // subscriber list and remove it from the database.
 274         (*p)->destroy();
 275         _subscribers.erase(p);
 276     }
 277
 278     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 279     _subscribers.push_back(subscriber);
 280 }
 281
 282 Ice::ObjectPrx
 283 TransientTopicImpl::subscribeAndGetPublisher(const QoS& qos, const Ice::ObjectPrx& obj, const Ice::Current&)
 284 {
 285     Ice::Identity id = obj->ice_getIdentity();
 286
 287     TraceLevelsPtr traceLevels = _instance->traceLevels();
 288     if(traceLevels->topic > 0)
 289     {
 290         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 291         out << _name << ": subscribeAndGetPublisher: " << _instance->communicator()->identityToString(id);
 292
 293         if(traceLevels->topic > 1)
 294         {
 295             out << " endpoints: " << IceStormInternal::describeEndpoints(obj)
 296                 << " QoS: ";
 297             for(QoS::const_iterator p = qos.begin(); p != qos.end() ; ++p)
 298             {
 299                 if(p != qos.begin())
 300                 {
 301                     out << ',';
 302                 }
 303
 304             }
 305         }
 306     }
 307
 308     Lock sync(*this);
 309
 310     SubscriberRecord record;
 311     record.id = id;
 312     record.obj = obj;
 313     record.theQoS = qos;
 314     record.topicName = _name;
 315     record.link = false;
 316     record.cost = 0;
 317
 318     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 319     if(p != _subscribers.end())
 320     {
 321         throw AlreadySubscribed();
 322     }
 323
 324     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 325     _subscribers.push_back(subscriber);
 326
 327     return subscriber->proxy();
 328 }
 329
 330 void
 331 TransientTopicImpl::unsubscribe(const Ice::ObjectPrx& subscriber, const Ice::Current&)
 332 {
 333     TraceLevelsPtr traceLevels = _instance->traceLevels();
 334     if(!subscriber)
 335     {
 336         if(traceLevels->topic > 0)
 337         {
 338             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 339             out << "unsubscribe with null subscriber.";
 340         }
 341         return;
 342     }
 343
 344     Ice::Identity id = subscriber->ice_getIdentity();
 345
 346     if(traceLevels->topic > 0)
 347     {
 348         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 349         out << _name << ": unsubscribe: " << _instance->communicator()->identityToString(id);
 350         if(traceLevels->topic > 1)
 351         {
 352             out << " endpoints: " << IceStormInternal::describeEndpoints(subscriber);
 353         }
 354     }
 355
 356     Lock sync(*this);
 357     // First remove the subscriber from the subscribers list. Note
 358     // that its possible that the subscriber isn't in the list, but is
 359     // in the database if the subscriber was locally reaped.
 360     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), id);
 361     if(p != _subscribers.end())
 362     {
 363         (*p)->destroy();
 364         _subscribers.erase(p);
 365     }
 366 }
 367
 368 TopicLinkPrx
 369 TransientTopicImpl::getLinkProxy(const Ice::Current&)
 370 {
 371     // immutable
 372     return _linkPrx;
 373 }
 374
 375 void
 376 TransientTopicImpl::link(const TopicPrx& topic, Ice::Int cost, const Ice::Current&)
 377 {
 378     TopicInternalPrx internal = TopicInternalPrx::uncheckedCast(topic);
 379     TopicLinkPrx link = internal->getLinkProxy();
 380
 381     TraceLevelsPtr traceLevels = _instance->traceLevels();
 382     if(traceLevels->topic > 0)
 383     {
 384         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 385         out << _name << ": link " << _instance->communicator()->identityToString(topic->ice_getIdentity())
 386             << " cost " << cost;
 387     }
 388
 389     Lock sync(*this);
 390
 391     Ice::Identity id = topic->ice_getIdentity();
 392
 393     SubscriberRecord record;
 394     record.id = id;
 395     record.obj = link;
 396     record.theTopic = topic;
 397     record.topicName = _name;
 398     record.link = true;
 399     record.cost = cost;
 400
 401     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), record.id);
 402     if(p != _subscribers.end())
 403     {
 404         string name = identityToTopicName(id);
 405         LinkExists ex;
 406         ex.name = name;
 407         throw ex;
 408     }
 409
 410     SubscriberPtr subscriber = Subscriber::create(_instance, record);
 411     _subscribers.push_back(subscriber);
 412 }
 413
 414 void
 415 TransientTopicImpl::unlink(const TopicPrx& topic, const Ice::Current&)
 416 {
 417     Lock sync(*this);
 418     if(_destroyed)
 419     {
 420         throw Ice::ObjectNotExistException(__FILE__, __LINE__);
 421     }
 422
 423     Ice::Identity id = topic->ice_getIdentity();
 424
 425     vector<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), id);
 426     if(p == _subscribers.end())
 427     {
 428         string name = identityToTopicName(id);
 429         TraceLevelsPtr traceLevels = _instance->traceLevels();
 430         if(traceLevels->topic > 0)
 431         {
 432             Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 433             out << _name << ": unlink " << name << " failed - not linked";
 434         }
 435
 436         NoSuchLink ex;
 437         ex.name = name;
 438         throw ex;
 439     }
 440
 441     TraceLevelsPtr traceLevels = _instance->traceLevels();
 442     if(traceLevels->topic > 0)
 443     {
 444         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 445         out << _name << " unlink " << _instance->communicator()->identityToString(id);
 446     }
 447
 448     // Remove the subscriber from the subscribers list. Note
 449     // that its possible that the subscriber isn't in the list, but is
 450     // in the database if the subscriber was locally reaped.
 451     p = find(_subscribers.begin(), _subscribers.end(), id);
 452     if(p != _subscribers.end())
 453     {
 454         (*p)->destroy();
 455         _subscribers.erase(p);
 456     }
 457 }
 458
 459 LinkInfoSeq
 460 TransientTopicImpl::getLinkInfoSeq(const Ice::Current&) const
 461 {
 462     Lock sync(*this);
 463     LinkInfoSeq seq;
 464     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
 465     {
 466         SubscriberRecord record = (*p)->record();
 467         if(record.link && !(*p)->errored())
 468         {
 469             LinkInfo info;
 470             info.name = identityToTopicName(record.theTopic->ice_getIdentity());
 471             info.cost = record.cost;
 472             info.theTopic = record.theTopic;
 473             seq.push_back(info);
 474         }
 475     }
 476     return seq;
 477 }
 478
 479 void
 480 TransientTopicImpl::destroy(const Ice::Current&)
 481 {
 482     Lock sync(*this);
 483
 484     if(_destroyed)
 485     {
 486         throw Ice::ObjectNotExistException(__FILE__, __LINE__);
 487     }
 488     _destroyed = true;
 489
 490     TraceLevelsPtr traceLevels = _instance->traceLevels();
 491     if(traceLevels->topic > 0)
 492     {
 493         Ice::Trace out(traceLevels->logger, traceLevels->topicCat);
 494         out << _name << ": destroy";
 495     }
 496
 497     try
 498     {
 499         _instance->publishAdapter()->remove(_linkPrx->ice_getIdentity());
 500         _instance->publishAdapter()->remove(_publisherPrx->ice_getIdentity());
 501     }
 502     catch(const Ice::ObjectAdapterDeactivatedException&)
 503     {
 504         // Ignore -- this could occur on shutdown.
 505     }
 506
 507     // Destroy all of the subscribers.
 508     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
 509     {
 510         (*p)->destroy();
 511     }
 512     _subscribers.clear();
 513 }
 514
 515 void
 516 TransientTopicImpl::reap(const Ice::IdentitySeq&, const Ice::Current&)
 517 {
 518 }
 519
 520 bool
 521 TransientTopicImpl::destroyed() const
 522 {
 523     Lock sync(*this);
 524     return _destroyed;
 525 }
 526
 527 Ice::Identity
 528 TransientTopicImpl::id() const
 529 {
 530     // immutable
 531     return _id;
 532 }
 533
 534 void
 535 TransientTopicImpl::publish(bool forwarded, const EventDataSeq& events)
 536 {
 537     //
 538     // Copy of the subscriber list so that event publishing can occur
 539     // in parallel.
 540     //
 541     vector<SubscriberPtr> copy;
 542     {
 543         Lock sync(*this);
 544         copy = _subscribers;
 545     }
 546
 547     //
 548     // Queue each event, gathering a list of those subscribers that
 549     // must be reaped.
 550     //
 551     vector<Ice::Identity> e;
 552     for(vector<SubscriberPtr>::const_iterator p = copy.begin(); p != copy.end(); ++p)
 553     {
 554         if(!(*p)->queue(forwarded, events) && (*p)->reap())
 555         {
 556             e.push_back((*p)->id());
 557         }
 558     }
 559
 560     //
 561     // Run through the error list removing those subscribers that are
 562     // in error from the subscriber list.
 563     //
 564     if(!e.empty())
 565     {
 566         Lock sync(*this);
 567         for(vector<Ice::Identity>::const_iterator ep = e.begin(); ep != e.end(); ++ep)
 568         {
 569             //
 570             // Its possible for the subscriber to already have been
 571             // removed since the copy is iterated over outside of
 572             // mutex protection.
 573             //
 574             // Note that although this could be quicker if we used a
 575             // map, the most optimal case should be pushing around
 576             // events not searching for a particular subscriber.
 577             //
 578             // The subscriber is immediately destroyed & removed from
 579             // the _subscribers list. Add the subscriber to a list of
 580             // error'd subscribers and remove it from the database on
 581             // the next reap.
 582             //
 583             vector<SubscriberPtr>::iterator q = find(_subscribers.begin(), _subscribers.end(), *ep);
 584             if(q != _subscribers.end())
 585             {
 586                 SubscriberPtr subscriber = *q;
 587                 //
 588                 // Destroy the subscriber.
 589                 //
 590                 subscriber->destroy();
 591                 _subscribers.erase(q);
 592             }
 593         }
 594     }
 595 }
 596
 597 void
 598 TransientTopicImpl::shutdown()
 599 {
 600     Lock sync(*this);
 601
 602     // Shutdown each subscriber. This waits for the event queues to drain.
 603     for(vector<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
 604     {
 605         (*p)->shutdown();
 606     }
 607 }