// Publish publishes a message.
327094
func (b *redisBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	v, err := b.opts.Codec.Marshal(msg)
	if err != nil {
		return err
	}

	conn := b.pool.Get()
	_, err = redis.Int(conn.Do("PUBLISH", topic, v))
	conn.Close()

	return err
}
1201226
func (e *Events) PublishMessage(jm eventtypes.Message) {
	eventsCounter.Inc()

	e.mu.Lock()
	if len(e.events) == cap(e.events) {
		// discard oldest event
		copy(e.events, e.events[1:])
		e.events[len(e.events)-1] = jm
	} else {
		e.events = append(e.events, jm)
	}
	e.mu.Unlock()
	e.pub.Publish(jm)
}
1239927
func (m *Driver) Publish(data interface{}) bool {
	message := data.([]byte)
	return m.adaptor().Publish(m.topic, message)
}
623353
func (nc *Conn) publish(subj, reply string, data []byte) error {
	if nc == nil {
		return ErrInvalidConnection
	}
	if subj == "" {
		return ErrBadSubject
	}
	nc.mu.Lock()

	if nc.isClosed() {
		nc.mu.Unlock()
		return ErrConnectionClosed
	}

	if nc.isDrainingPubs() {
		nc.mu.Unlock()
		return ErrConnectionDraining
	}

	// Proactively reject payloads over the threshold set by server.
	msgSize := int64(len(data))
	if msgSize > nc.info.MaxPayload {
		nc.mu.Unlock()
		return ErrMaxPayload
	}

	// Check if we are reconnecting, and if so check if
	// we have exceeded our reconnect outbound buffer limits.
	if nc.isReconnecting() {
		// Flush to underlying buffer.
		nc.bw.Flush()
		// Check if we are over
		if nc.pending.Len() >= nc.Opts.ReconnectBufSize {
			nc.mu.Unlock()
			return ErrReconnectBufExceeded
		}
	}

	msgh := nc.scratch[:len(_PUB_P_)]
	msgh = append(msgh, subj...)
	msgh = append(msgh, ' ')
	if reply != "" {
		msgh = append(msgh, reply...)
		msgh = append(msgh, ' ')
	}

	// We could be smarter here, but simple loop is ok,
	// just avoid strconv in fast path
	// FIXME(dlc) - Find a better way here.
	// msgh = strconv.AppendInt(msgh, int64(len(data)), 10)

	var b [12]byte
	var i = len(b)
	if len(data) > 0 {
		for l := len(data); l > 0; l /= 10 {
			i -= 1
			b[i] = digits[l%10]
		}
	} else {
		i -= 1
		b[i] = digits[0]
	}

	msgh = append(msgh, b[i:]...)
	msgh = append(msgh, _CRLF_...)

	_, err := nc.bw.Write(msgh)
	if err == nil {
		_, err = nc.bw.Write(data)
	}
	if err == nil {
		_, err = nc.bw.WriteString(_CRLF_)
	}
	if err != nil {
		nc.mu.Unlock()
		return err
	}

	nc.OutMsgs++
	nc.OutBytes += uint64(len(data))

	if len(nc.fch) == 0 {
		nc.kickFlusher()
	}
	nc.mu.Unlock()
	return nil
}
1226178
func (w *Producer) Publish(topic string, body []byte) error {
	return w.sendCommand(Publish(topic, body))
}
797943
func (p *PubSub) Publish(i interface{}) {
	p.m.Lock()
	defer p.m.Unlock()

	for _, v := range p.subscribers {
		v.OnEvent(i)
	}
}
426063
func (c *Client) Publish(topic string, args []interface{}, kwargs map[string]interface{}) error {
	return c.Send(&Publish{
		Request:     NewID(),
		Options:     make(map[string]interface{}),
		Topic:       URI(topic),
		Arguments:   args,
		ArgumentsKw: kwargs,
	})
}
429457
func (em *Emitter) Publish(ctx context.Context, message interface{}, tags query.Tagged) error {
	if em == nil || em.pubsubServer == nil {
		return nil
	}
	return em.pubsubServer.PublishWithTags(ctx, message, tags)
}
687243
func (queue *redisQueue) Publish(payload string) bool {
	// debug(fmt.Sprintf("publish %s %s", payload, queue)) // COMMENTOUT
	return queue.redisClient.LPush(queue.readyKey, payload)
}
1105141
func (e *eventer) Publish(name string, data interface{}) {
	evt := NewEvent(name, data)
	e.in <- evt
}
190568
func (pcc *PubControlClient) Publish(channel string, item *Item) error {
	return pcc.publish(pcc, channel, item)
}
230656
func (mr *MockClientConnectorMockRecorder) Publish(msg interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockClientConnector)(nil).Publish), msg)
}
101407
func (np NoopProvider) Publish(ctx context.Context, topic string, m *Msg) error {
	return nil
}
546903
func (r *remoteServer) Publish(message *params.PubSubMessage) {
	select {
	case <-r.tomb.Dying():
		r.logger.Tracef("dying, don't send %q", message.Topic)
	default:
		r.mutex.Lock()
		// Only queue the message up if we are currently connected.
		notifyData := false
		if r.connection != nil {
			r.logger.Tracef("queue up topic %q", message.Topic)
			r.pending.PushBack(message)
			notifyData = r.pending.Len() == 1

		} else {
			r.logger.Tracef("skipping %q for %s as not connected", message.Topic, r.target)
		}
		r.mutex.Unlock()
		if notifyData {
			select {
			case r.data <- struct{}{}:
			case <-r.connectionReset:
				r.logger.Debugf("connection reset while notifying %q for %s", message.Topic, r.target)
			}
		}
	}
}
1211264
func (s *Subscriber) Publish(c, msg string) int {
	s.mu.Lock()
	defer s.mu.Unlock()

	found := 0

subs:
	for sub := range s.channels {
		if sub == c {
			s.publish <- PubsubMessage{c, msg}
			found++
			break subs
		}
	}

pats:
	for _, pat := range s.patterns {
		if pat.MatchString(c) {
			s.publish <- PubsubMessage{c, msg}
			found++
			break pats
		}
	}

	return found
}
1107020
func (c *ChannelDeleteCall) Publish(publish bool) *ChannelDeleteCall {
	c.urlParams_.Set("publish", fmt.Sprint(publish))
	return c
}
50198
func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error {
	// Adjust routing key (this decides which queue the message will be published to)
	b.AdjustRoutingKey(signature)

	msg, err := json.Marshal(signature)
	if err != nil {
		return fmt.Errorf("JSON marshal error: %s", err)
	}

	// Check the ETA signature field, if it is set and it is in the future,
	// delay the task
	if signature.ETA != nil {
		now := time.Now().UTC()

		if signature.ETA.After(now) {
			delayMs := int64(signature.ETA.Sub(now) / time.Millisecond)

			return b.delay(signature, delayMs)
		}
	}

	connection, err := b.GetOrOpenConnection(signature.RoutingKey,
		b.GetConfig().AMQP.BindingKey, // queue binding key
		nil,                           // exchange declare args
		nil,                           // queue declare args
		amqp.Table(b.GetConfig().AMQP.QueueBindingArgs), // queue binding args
	)
	if err != nil {
		return err
	}

	channel := connection.channel
	confirmsChan := connection.confirmation

	if err := channel.Publish(
		b.GetConfig().AMQP.Exchange, // exchange name
		signature.RoutingKey,        // routing key
		false,                       // mandatory
		false,                       // immediate
		amqp.Publishing{
			Headers:      amqp.Table(signature.Headers),
			ContentType:  "application/json",
			Body:         msg,
			DeliveryMode: amqp.Persistent,
		},
	); err != nil {
		return err
	}

	confirmed := <-confirmsChan

	if confirmed.Ack {
		return nil
	}

	return fmt.Errorf("Failed delivery of delivery tag: %v", confirmed.DeliveryTag)
}