// Publish places a new message on the default queue
551842
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.Broker.AdjustRoutingKey(signature)

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

	conn := b.open()
	defer conn.Close()

	// 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) {
			score := signature.ETA.UnixNano()
			_, err = conn.Do("ZADD", redisDelayedTasksKey, score, msg)
			return err
		}
	}

	_, err = conn.Do("RPUSH", signature.RoutingKey, msg)
	return err
}
941404
func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error {
	msg, err := json.Marshal(signature)
	if err != nil {
		return fmt.Errorf("JSON marshal error: %s", err)
	}

	// Check that signature.RoutingKey is set, if not switch to DefaultQueue
	b.AdjustRoutingKey(signature)

	MsgInput := &awssqs.SendMessageInput{
		MessageBody: aws.String(string(msg)),
		QueueUrl:    aws.String(b.GetConfig().Broker + "/" + signature.RoutingKey),
	}

	// if this is a fifo queue, there needs to be some additional parameters.
	if strings.HasSuffix(signature.RoutingKey, ".fifo") {
		// Use Machinery's signature Task UUID as SQS Message Group ID.
		MsgDedupID := signature.UUID
		MsgInput.MessageDeduplicationId = aws.String(MsgDedupID)

		// Do not Use Machinery's signature Group UUID as SQS Message Group ID, instead use BrokerMessageGroupId
		MsgGroupID := signature.BrokerMessageGroupId
		if MsgGroupID == "" {
			return fmt.Errorf("please specify BrokerMessageGroupId attribute for task Signature when submitting a task to FIFO queue")
		}
		MsgInput.MessageGroupId = aws.String(MsgGroupID)
	}

	// Check the ETA signature field, if it is set and it is in the future,
	// and is not a fifo queue, set a delay in seconds for the task.
	if signature.ETA != nil && !strings.HasSuffix(signature.RoutingKey, ".fifo") {
		now := time.Now().UTC()
		delay := signature.ETA.Sub(now)
		if delay > 0 {
			if delay > maxAWSSQSDelay {
				return errors.New("Max AWS SQS delay exceeded")
			}
			MsgInput.DelaySeconds = aws.Int64(int64(delay.Seconds()))
		}
	}

	result, err := b.service.SendMessageWithContext(ctx, MsgInput)

	if err != nil {
		log.ERROR.Printf("Error when sending a message: %v", err)
		return err

	}
	log.INFO.Printf("Sending a message successfully, the messageId is %v", *result.MessageId)
	return nil

}
749485
func (c *Client) Publish(topic string, payload []byte, qos packet.QOS, retain bool) (GenericFuture, error) {
	msg := &packet.Message{
		Topic:   topic,
		Payload: payload,
		QOS:     qos,
		Retain:  retain,
	}

	return c.PublishMessage(msg)
}
228363
func (c *Client) Publish(pub *Publisher) {
	c.l.Lock()
	defer c.l.Unlock()
	c.publishers[pub] = struct{}{}
	if ch, err := c.channel(); err == nil {
		go pub.serve(c, ch)
	}
}
92879
func (service *BaseService) Publish(
	topic string,
	timeout time.Duration,
	heartbeat time.Duration) Service {
	if timeout <= 0 {
		timeout = service.Timeout
	}
	if heartbeat <= 0 {
		heartbeat = service.Heartbeat
	}
	t := newTopic(heartbeat)
	service.topicLock.Lock()
	service.topics[topic] = t
	service.topicLock.Unlock()
	return service.AddFunction(topic, func(id string) interface{} {
		message := t.get(id)
		if message == nil {
			message = make(chan interface{})
			t.put(id, message)
			fireSubscribeEvent(topic, id, service)
		}
	receiveMessage:
		select {
		case result := <-message:
			if result == heartbeatSignals {
				goto receiveMessage
			}
			return result
		case <-time.After(timeout):
			go func() {
				select {
				case message <- heartbeatSignals:
					break
				case <-time.After(t.heartbeat):
					service.offline(t, topic, id)
				}
			}()
			return nil
		}
	}, Options{})
}
74126
func (p *NatsPublisher) Publish(subject string, r io.Reader) error {
	_, err := p.server.initSubject(subject)
	p.server.queue[subject] <- r
	return err
}
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
}
669427
func (ps *PubSub) Publish(topic string, item interface{}) error {
	ps.RLock()
	defer ps.RUnlock()
	s, subscribed := ps.subscriptions[topic]
	if !subscribed {
		return errors.New("no subscribers")
	}
	for _, sub := range s.ToArray() {
		c := sub.(*subscription).c
		// Not enough room in buffer, continue in order to not block publisher
		if len(c) == subscriptionBuffSize {
			continue
		}
		c <- item
	}
	return nil
}
114305
func (b *sqsBroker) Publish(queueName string, msg *broker.Message, opts ...broker.PublishOption) error {
	queueURL, err := b.urlFromQueueName(queueName)
	if err != nil {
		return err
	}

	input := &sqs.SendMessageInput{
		MessageBody: aws.String(string(msg.Body[:])),
		QueueUrl:    &queueURL,
	}
	input.MessageAttributes = copyMessageHeader(msg)
	input.MessageDeduplicationId = b.generateDedupID(msg)
	input.MessageGroupId = b.generateGroupID(msg)

	log.Log(fmt.Sprintf("Publishing SQS message, %d bytes", len(msg.Body)))
	_, err = b.svc.SendMessage(input)

	if err != nil {
		return err
	}

	// Broker interfaces don't let us do anything with message ID or sequence number
	return nil
}
325053
func (Pipe) Publish(ctx *context.Context) error {
	if len(ctx.Config.S3) == 0 {
		return pipe.Skip("s3 section is not configured")
	}
	var g = semerrgroup.New(ctx.Parallelism)
	for _, conf := range ctx.Config.S3 {
		conf := conf
		g.Go(func() error {
			return upload(ctx, conf)
		})
	}
	return g.Wait()
}
1250017
func (m *Miniredis) Publish(channel, message string) int {
	m.Lock()
	defer m.Unlock()
	return m.publish(channel, message)
}
895156
func (b *pubsubBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	if b.err != nil {
		return b.err
	}
	t := b.topic(topic)
	return t.Send(context.Background(), &pubsub.Message{Metadata: msg.Header, Body: msg.Body})
}
1030662
func (c *Client) Publish(channel, message string) error {
	_, err := c.execWithKey(true, "PUBLISH", channel, message)
	return err
}
892340
func (r *Redis) Publish(channel, message string) (int64, error) {
	rp, err := r.ExecuteCommand("PUBLISH", channel, message)
	if err != nil {
		return 0, err
	}
	return rp.IntegerValue()
}
828635
func (client *Client) Publish(channel string, val []byte) error {
    _, err := client.sendCommand("PUBLISH", channel, string(val))
    if err != nil {
        return err
    }
    return nil
}
671370
func (b *pubsubBroker) Publish(topic string, msg *broker.Message, opts ...broker.PublishOption) error {
	t := b.client.Topic(topic)
	ctx := context.Background()

	exists, err := t.Exists(ctx)
	if err != nil {
		return err
	}

	if !exists {
		tt, err := b.client.CreateTopic(ctx, topic)
		if err != nil {
			return err
		}
		t = tt
	}

	m := &pubsub.Message{
		ID:         "m-" + uuid.New().String(),
		Data:       msg.Body,
		Attributes: msg.Header,
	}

	pr := t.Publish(ctx, m)
	_, err = pr.Get(ctx)
	return err
}
711512
func (mr *MockInstanceConnectorMockRecorder) Publish(msg interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockInstanceConnector)(nil).Publish), msg)
}