void VisionSystem::pipeLineDetectorOn()
{
    addDownwardDetector(m_pipelineDetector);
    publish(EventType::PIPELINE_DETECTOR_ON,
            core::EventPtr(new core::Event()));
}

void Telemetry::pub(const char * topic, const char * msg)
{
    publish(topic,msg);
}

bool SynapseSimulatorUpdater<MType>::
start(std::function<bool()> thread_init,
      std::function<bool()> thread_destroy) {
  struct Synchronizer : public DataBuffer {
    DeviceNeuronStateBuffer<MType>* neuron_state;
    SynapticFireVectorBuffer<MType>* synaptic_fire;
    SynapticCurrentBuffer<MType>* synaptic_current;
    float simulation_time;
    float time_step;
  };
  auto synchronizer_publisher = new SpecificPublisher<Synchronizer>();
  for (size_t i = 0; i < num_buffers_; ++i) {
    synchronizer_publisher->addBlank(new Synchronizer());
  }
  auto master_function = [this,
                          synchronizer_publisher,
                          thread_init,
                          thread_destroy]() {
    thread_init();
    float simulation_time = 0.0f;
    float time_step = simulation_parameters_->getTimeStep();
    unsigned int simulation_step = 0;
    Mailbox mailbox;
    while(true) {
      DeviceNeuronStateBuffer<MType>* neuron_state = nullptr;
      neuron_state_subscription_->pull(&neuron_state, &mailbox);
      SynapticFireVectorBuffer<MType>* synaptic_fire = nullptr;
      fire_subscription_->pull(&synaptic_fire, &mailbox);
      if (!mailbox.wait(&neuron_state, &synaptic_fire)) {
        neuron_state_subscription_->cancel();
        fire_subscription_->cancel();
        if (neuron_state) {
          neuron_state->release();
        }
        if (synaptic_fire) {
          synaptic_fire->release();
        }
        delete synchronizer_publisher;
        break;
      }
      auto synchronizer = synchronizer_publisher->getBlank();
      auto synaptic_current = this->getBlank();
      synaptic_current->simulation_step = simulation_step;
      if (!synaptic_current->clear()) {
        std::cerr << "Failed to clear SynapticCurrentBuffer." <<
          std::endl;
      }
      synchronizer->neuron_state = neuron_state;
      synchronizer->synaptic_fire = synaptic_fire;
      synchronizer->synaptic_current = synaptic_current;
      synchronizer->time_step = time_step;
      synchronizer->simulation_time = simulation_time;
      auto prerelease_function = [this, synchronizer]() {
        this->publish(synchronizer->synaptic_current);
        synchronizer->neuron_state->release();
        synchronizer->synaptic_fire->release();
      };
      synchronizer->setPrereleaseFunction(prerelease_function);
      synchronizer_publisher->publish(synchronizer);
      simulation_time += time_step;
      ++simulation_step;
    }
    thread_destroy();
  };
  master_thread_ = std::thread(master_function);

  for (size_t i = 0; i < simulators_.size(); ++i) {
    auto simulator = simulators_[i];
    auto unit_offset = device_synaptic_vector_offsets_[i];
    auto subscription = synchronizer_publisher->subscribe();
    auto worker_function = [subscription, 
                            simulator, 
                            unit_offset,
                            thread_init,
                            thread_destroy]() {
      thread_init();
      auto word_offset = Bit::num_words(unit_offset);
      while (true) {
        auto synchronizer = subscription->pull();
        if (nullptr == synchronizer) {
          delete subscription;
          break;
        }
        SynapseUpdateParameters parameters;
        parameters.synaptic_fire =
          synchronizer->synaptic_fire->getFireBits() + word_offset;
        parameters.neuron_voltage =
          synchronizer->neuron_state->getVoltages();
        parameters.device_neuron_fire =
          synchronizer->neuron_state->getFireBits();
        parameters.synaptic_current =
          synchronizer->synaptic_current->getCurrents();
        parameters.write_lock =
          synchronizer->synaptic_current->getWriteLock();
        parameters.simulation_time = synchronizer->simulation_time;
        parameters.time_step = synchronizer->time_step;
        if (!simulator->update(&parameters)) {
          std::cerr << "An error occurred updating a SynapseSimulator." <<
            std::endl;
        }
//.........这里部分代码省略.........

void rice::tutorial::splitstream::MySplitStreamClient::deliver(::rice::p2p::commonapi::Id* id, ::rice::p2p::commonapi::Message* message)
{
    if(dynamic_cast< MySplitStreamClient_PublishContent* >(message) != nullptr) {
        publish();
    }
}

 void metadataSubscriptionCallback(ros::pub_sub_conn* conn)
 {
   publish( "map_metadata", meta_data_message_ );
 }

void VisionSystem::hedgeDetectorOn()
{
    addForwardDetector(m_hedgeDetector);
    publish(EventType::HEDGE_DETECTOR_ON,
            core::EventPtr(new core::Event()));
}

bool MQTT::publish(String& topic, const char* buf, uint32_t buf_len, int qos, int retain)
{
	return publish(topic.c_str(), buf, buf_len, qos, retain);
}

void VisionSystem::downwardSafeDetectorOff()
{
    m_downward->removeDetector(m_downwardSafeDetector);
    publish(EventType::SAFE_DETECTOR_OFF,
            core::EventPtr(new core::Event()));
}

bool MQTTclient::publish(const String& topic, const float& msg) {
	return publish(topic, String(msg));
}

void MosqConnect::pub(QString topic, QString subject)
{
    publish(NULL, topic.toAscii(), subject.size(), subject.toAscii());
}

String MessageChannel::publish( MessageSession *p_session , MessagePublisher *pub , const char *msg ) {
	TextMessage *l_msg = new TextMessage();
	l_msg -> setMessageType( pub -> msgtype );
	l_msg -> setText( msg );
	return( publish( p_session , pub , l_msg ) );
}

/*!
 * \brief Publish a message
 * \param topic The topic of the publish
 * \param msg The (ASCII) message to publish
 *
 * Create a PUBLISH packet and send it to the MQTT server
 *
 * \returns false if sending the message failed somehow
 */
bool MQTT::publish(const char * topic, const char * msg, uint8_t qos)
{
    return publish(topic, (const uint8_t *)msg, strlen(msg), qos);
}