publishes lines one by one to the given topic
971368
def callback(self, filename, lines, **kwargs):
        """"""
        timestamp = self.get_timestamp(**kwargs)
        if kwargs.get('timestamp', False):
            del kwargs['timestamp']

        for line in lines:
            try:
                import warnings
                with warnings.catch_warnings():
                    warnings.simplefilter('error')
                    #produce message
                    if self._key is None:
                        response = self._prod.send_messages(self._kafka_config['topic'], self.format(filename, line, timestamp, **kwargs))
                    else:
                        response = self._prod.send_messages(self._kafka_config['topic'], self._key, self.format(filename, line, timestamp, **kwargs))

                    if response:
                        if response[0].error:
                            self._logger.info('message error: {0}'.format(response[0].error))
                            self._logger.info('message offset: {0}'.format(response[0].offset))

            except Exception as e:
                try:
                    self._logger.error('Exception caught sending message/s : ' + str(e))
                    raise TransportException(e.strerror)
                except AttributeError:
                    raise TransportException('Unspecified exception encountered')
14013
function serializeTopics(topics) {
    return topics.map(function (topic) {
        if (typeof (topic) === 'string') {
            return topic;
        }
        else if (Array.isArray(topic)) {
            topic.forEach(function (topic) {
                if (topic !== null && bytes_1.hexDataLength(topic) !== 32) {
                    errors.throwError('invalid topic', errors.INVALID_ARGUMENT, { argument: 'topic', value: topic });
                }
            });
            return topic.join(',');
        }
        else if (topic === null) {
            return '';
        }
        return errors.throwError('invalid topic value', errors.INVALID_ARGUMENT, { argument: 'topic', value: topic });
    }).join('&');
}
737139
def publish(self):
        

        try:
             for collection in self.settings.get("scheduler").get("collections"):
                yield self.publish_for_collection(collection)
        except Exception as ex:
            self.logger.error(ex)
720801
def publish(self, titles):
        
        if isinstance(titles, Sequence) and not isinstance(titles, six.string_types):
            for title in titles:
                self.episodes[title].publish()
        elif isinstance(titles, six.string_types):
            self.episodes[titles].publish()
        else:
            raise TypeError('titles must be a string or a sequence of strings.')

        self.update_rss_feed()
193734
private void generateCopies(final Element topicref, final List<FilterUtils> filters) {
        final List<FilterUtils> fs = combineFilterUtils(topicref, filters);

        final String copyTo = topicref.getAttribute(BRANCH_COPY_TO);
        if (!copyTo.isEmpty()) {
            final URI dstUri = map.resolve(copyTo);
            final URI dstAbsUri = job.tempDirURI.resolve(dstUri);
            final String href = topicref.getAttribute(ATTRIBUTE_NAME_HREF);
            final URI srcUri = map.resolve(href);
            final URI srcAbsUri = job.tempDirURI.resolve(srcUri);
            final FileInfo srcFileInfo = job.getFileInfo(srcUri);
            if (srcFileInfo != null) {
//                final FileInfo fi = new FileInfo.Builder(srcFileInfo).uri(dstUri).build();
//                 TODO: Maybe Job should be updated earlier?
//                job.add(fi);
                logger.info("Filtering " + srcAbsUri + " to " + dstAbsUri);
                final ProfilingFilter writer = new ProfilingFilter();
                writer.setLogger(logger);
                writer.setJob(job);
                writer.setFilterUtils(fs);
                writer.setCurrentFile(dstAbsUri);
                final List<XMLFilter> pipe = singletonList(writer);

                final File dstDirUri = new File(dstAbsUri.resolve("."));
                if (!dstDirUri.exists() && !dstDirUri.mkdirs()) {
                    logger.error("Failed to create directory " + dstDirUri);
                }
                try {
                    xmlUtils.transform(srcAbsUri,
                                       dstAbsUri,
                                       pipe);
                } catch (final DITAOTException e) {
                    logger.error("Failed to filter " + srcAbsUri + " to " + dstAbsUri + ": " + e.getMessage(), e);
                }
                topicref.setAttribute(ATTRIBUTE_NAME_HREF, copyTo);
                topicref.removeAttribute(BRANCH_COPY_TO);
                // disable filtering again
                topicref.setAttribute(SKIP_FILTER, Boolean.TRUE.toString());
            }
        }
        for (final Element child: getChildElements(topicref, MAP_TOPICREF)) {
            if (DITAVAREF_D_DITAVALREF.matches(child)) {
                continue;
            }
            generateCopies(child, fs);
        }
    }
1058677
def pub(self, topic, message):
        ''''''
        with self.random_connection() as client:
            client.pub(topic, message)
            return self.wait_response()
906182
def write_lines(self, lines, level=0):
        """"""

        for line in lines:
            self.write_line(line, level)
619175
def publish(self,topic,options=None,args=None,kwargs=None):
        
        topic = self.get_full_uri(topic)
        if options is None:
            options = {'acknowledge':True}
        if options.get('acknowledge'):
            request = PUBLISH(
                        options=options or {},
                        topic=topic,
                        args=args or [],
                        kwargs=kwargs or {}
                      )
            result = self.send_and_await_response(request)
            return result
        else:
            request = PUBLISH(
                        options=options or {},
                        topic=topic,
                        args=args or [],
                        kwargs=kwargs or {}
                      )
            self.send_message(request)
            return request.request_id
1341194
public void publishInternal(Object message) {
        topicStats.incrementPublishes();
        topicService.publishMessage(name, message, multithreaded);
    }
1334153
def publish(self, message, key=None, **kws):
        """"""
        return PSPut(self, message, key, **kws)
1229259
def mpub(self, topic, messages, binary=True):
        ''''''
        if binary:
            # Pack and ship the data
            return self.post('mpub', data=pack(messages)[4:],
                params={'topic': topic, 'binary': True})
        elif any('\n' in m for m in messages):
            # If any of the messages has a newline, then you must use the binary
            # calling format
            raise ClientException(
                'Use `binary` flag in mpub for messages with newlines')
        else:
            return self.post(
                '/mpub', params={'topic': topic}, data='\n'.join(messages))
1019571
def _send(self):
        

        while len(self.queue) > 0:
            metric = self.queue.popleft()

            path = '%s.%s.%s' % (
                metric.getPathPrefix(),
                metric.getCollectorPath(),
                metric.getMetricPath()
            )

            topic, value, timestamp = str(metric).split()
            logging.debug(
                "Sending.. topic[%s], value[%s], timestamp[%s]",
                path,
                value,
                timestamp
            )

            self.api.metric(path, (timestamp, value), host=metric.host)
44162
def publish(topic, args=nil, kwargs=nil, options={}, &callback)
      command = WampRails::Command::Publish.new(topic, args, kwargs, options, self)
      self._queue_command(command, callback)
    end
456966
@RobotKeywordOverload
    @ArgumentNames({ "topic", "message" })
    public void publishToMQTTSynchronously(String topic, Object message)
            throws MqttException {
        publishToMQTTSynchronously(topic, message, 0, false);
    }
932347
private void publishMultiTaskData(WorkUnitState state, int branchId, Set<Path> writerOutputPathsMoved)
      throws IOException {
    publishData(state, branchId, false, writerOutputPathsMoved);
    addLineageInfo(state, branchId);
  }
932997
public void broadcastTextMessage(String topicName, String textMessage, int delaySeconds)
            throws NamingException, JMSException, ServiceLocatorException {
        if (mdwMessageProducer != null) {
            mdwMessageProducer.broadcastMessageToTopic(topicName, textMessage);
        }
        else {
            TopicConnectionFactory tFactory = null;
            TopicConnection tConnection = null;
            TopicSession tSession = null;
            TopicPublisher tPublisher = null;
            try {
                // if (logger.isDebugEnabled()) logger.debug("broadcast JMS
                // message: " +
                // textMessage);
                // cannot log above - causing recursive broadcasting
                tFactory = getTopicConnectionFactory(null);
                tConnection = tFactory.createTopicConnection();
                tSession = tConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
                Topic topic = getTopic(topicName);
                tPublisher = tSession.createPublisher(topic);

                // TODO: platform-independent delay
                // WLMessageProducer wlMessageProducer =
                // (WLMessageProducer)tPublisher;
                // long delayInMilliSec = 0;
                // if(pMinDelay > 0){
                // delayInMilliSec = delaySeconds*1000;
                // }
                // wlMessageProducer.setTimeToDeliver(delayInMilliSec);
                TextMessage message = tSession.createTextMessage();
                tConnection.start();

                message.setText(textMessage);
                tPublisher.publish(message, DeliveryMode.PERSISTENT,
                        TextMessage.DEFAULT_DELIVERY_MODE, TextMessage.DEFAULT_TIME_TO_LIVE);
                // }catch(ServiceLocatorException ex){
                // ex.printStackTrace();
                // never log exception here!!! infinite loop when publishing log
                // messages
                // throw new JMSException(ex.getMessage());
            }
            finally {
                closeResources(tConnection, tSession, tPublisher);

            }
        }
    }
96826
def put_content(self, content):
        
        if not isinstance(content, str):
            content = content.decode('utf-8')

        self.topic.publish(Message=content, **self.storage_args)