Process events and publish data
766118
def process(self, salt_data, token, opts):
        '''
        
        '''
        log.debug('In process %s', threading.current_thread())
        log.debug(salt_data['tag'])
        log.debug(salt_data)

        parts = salt_data['tag'].split('/')
        if len(parts) < 2:
            return

        # TBD: Simplify these conditional expressions
        if parts[1] == 'job':
            log.debug('In job part 1')
            if parts[3] == 'new':
                log.debug('In new job')
                self.process_new_job_event(salt_data)
                # if salt_data['data']['fun'] == 'grains.items':
                #     self.minions = {}
            elif parts[3] == 'ret':
                log.debug('In ret')
                self.process_ret_job_event(salt_data)
                if salt_data['data']['fun'] == 'grains.items':
                    self.process_minion_update(salt_data)
        elif parts[1] == 'key':
            log.debug('In key')
            self.process_key_event(salt_data)
        elif parts[1] == 'presence':
            self.process_presence_events(salt_data, token, opts)
64456
async def emit_event(self, event):
        
        self.log.info("publishing event on %s", self.publish_topic)
        if self.config.extra['config']['pub_options']['retain']:
            try:
                await persist_event(
                    self.publish_topic,
                    event,
                    self.pool
                )
            except SystemError as error:
                self.log.error(error)
                return

        loop = asyncio.get_event_loop()
        producer = AIOKafkaProducer(
            loop=loop,
            bootstrap_servers=self.transport_host
        )
        await producer.start()

        try:
            event = json.dumps(event.__dict__).encode()
            await producer.send_and_wait(
                self.publish_topic,
                event
            )
        finally:
            await producer.stop()
327619
public void fireEvent(Object publisher, FireAfterTransaction events) throws RuntimeException {
        try {
            for (int i = 0; i < events.events().length; i++) {
                Class<? extends EventObject> event = events.events()[i];
                if (ApplicationEvent.class.isAssignableFrom(event)) {
                    ctx.publishEvent((ApplicationEvent) event.getConstructor(Object.class).newInstance(publisher));
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
1060895
protected void publishResourceEvent(
            IPortletWindow portletWindow,
            HttpServletRequest httpServletRequest,
            long executionTime,
            boolean usedBrowserCache,
            boolean usedPortalCache) {

        final IPortletWindowId portletWindowId = portletWindow.getPortletWindowId();

        this.portalEventFactory.publishPortletResourceExecutionEvent(
                httpServletRequest,
                this,
                portletWindowId,
                executionTime,
                usedBrowserCache,
                usedPortalCache);
    }
180119
private static void publishEvent(String event, SiteNode node) {
		ZAP.getEventBus().publishSyncEvent(SiteMapEventPublisher.getPublisher(), new Event(SiteMapEventPublisher.getPublisher(), event, new Target(node)));
	}
1166403
private void publishEvent(WSJobExecution objectToPublish, String eventToPublishTo) {
        if (getBatchEventsPublisher() != null) {
            getBatchEventsPublisher().publishJobExecutionEvent(objectToPublish, eventToPublishTo, correlationId);
        }
    }
46969
public function publish(array $events, CommandBus $commandBus, bool $qeueEvents = false): void
    {
        $eventStreamObjects = [];
        /**
         * @var EventInterface $event
         */
        foreach ($events as $event) {
            // Save the event to the event stream.
            $eventStreamObject = new EventStreamObject();
            $eventStreamObject->setUuid($event->getCommand()->getAggregateUuid());
            $eventStreamObject->setCommandUuid($event->getCommand()->getUuid());
            $eventStreamObject->setPayload($event->getCommand()->getPayload());
            $eventStreamObject->setMessage($event->getMessage());
            $eventStreamObject->setEvent(\get_class($event));
            $eventStreamObject->setAggregateClass($event->getCommand()->getAggregateClass());
            $eventStreamObject->setVersion($event->getCommand()->getOnVersion() + 1);
            $eventStreamObject->setUser($event->getCommand()->getUser());

            // Add to list of eventStreamObjects so we can later notify aggregateSubscribers.
            $eventStreamObjects[] = $eventStreamObject;

            // Add the events to the eventStore.
            if ($qeueEvents) {
                $eventQeueObject = new EventQeueObject($eventStreamObject);
                $this->eventStore->qeue($eventQeueObject);
                $message = 'Qeued event: '.$event->getMessage();
            } else {
                $this->eventStore->add($eventStreamObject);
                $message = 'Persisted event: '.$event->getMessage();
            }

            // Dispatch messages about the events.
            $this->messageBus->dispatch(new Message(
                $message,
                $event::getCode(),
                $event->getCommand()->getUuid(),
                $event->getCommand()->getAggregateUuid(),
                null,
                $event->getCommand()->getPayload()
            ));
        }

        try {
            $this->eventStore->save();

            // Listeners are called even if the events are just qeued!
            // They are not called again when the events are persisted to the event stream!
            $this->invokeListeners($events, $commandBus);

            // Notify subscribers.
            // Subscribers are notified AFTER the events listeners are processed.
            // Subscribers are NOT notified of qeued events.
            if (!$qeueEvents) {
                $this->sendAggregateUpdates($eventStreamObjects);
            }
        } catch (\Exception $e) {
            // Saving to the Event Store failed. This can happen for example when an aggregate version is already taken.
            $this->messageBus->dispatch(new Message(
                $e->getMessage(),
                $e->getCode(),
                null,
                null,
                $e
            ));
        }
    }
1273806
private void publishStartedEvent() {
        BatchEventsPublisher publisher = getBatchEventsPublisher();
        if (publisher != null) {

            publisher.publishSplitFlowEvent(getSplitName(),
                                            getFlowName(),
                                            getTopLevelInstanceId(),
                                            getTopLevelExecutionId(),
                                            BatchEventsPublisher.TOPIC_EXECUTION_SPLIT_FLOW_STARTED,
                                            correlationId);
        }
    }
1151038
def publish(self, key, data):
        '''
        
        '''
        publish_data = {key: data}
        self.handler.send(salt.utils.json.dumps(publish_data), False)
1038504
function publishEvent( channelName, eventObj ) {
  var broker = getEventBroker( 'main' )
  eventObj.channel = channelName
  broker.publish( eventObj ); 
}
477753
def publish(self, data, **kwargs):
        """"""
        assert data.get('op') in {'index', 'create', 'delete', 'update'}
        return super(Producer, self).publish(data, **kwargs)
780570
function publish( data ) {
	var root;
	var docs;

	// Remove undocumented doclets:
	data( opts ).remove();

	// Retrieve an array of raw doclet objects:
	docs = data().get();

	// Transform the raw doclet array:
	root = {};
	transform( root, docs );

	// Write to stdout:
	console.log( JSON.stringify( root ) );
}
701389
def publish(self):
        """"""
        while not self.ws.closed:
            # Sleep to prevent *constant* context-switches.
            gevent.sleep(self.lag_tolerance_secs)
            message = self.ws.receive()
            if message is not None:
                channel_name, data = message.split(":", 1)
                redis_conn.publish(channel_name, data)
1209346
public void publishEvent(final ApplicationEvent event) {
        if (this.eventPublisher != null) {
            LOGGER.trace("Publishing event [{}]", event);
            this.eventPublisher.publishEvent(event);
        }
    }
462146
protected void doPublishEvent(final ApplicationEvent e) {
        if (applicationEventPublisher != null) {
            LOGGER.trace("Publishing [{}]", e);
            this.applicationEventPublisher.publishEvent(e);
        }
    }
893909
def publish (self):
        '''
        
        '''
        #print(self)
        #self.using_event.wait()

        self.lock.acquire()
        msg = cmdvel2PosTarget(self.vel)
        self.lock.release()
        self.pub.publish(msg)
677399
function dispatchEvent (event, data) {
    if (hasDOM()) {
      eventEmitter.dispatchEvent(new CustomEvent(event, data));
    } else if (hasEventEmitter()) {
      eventEmitter.emit(event, data);
    }
  }