// PublisherAfter sets the ClientResponseFuncs applied to the incoming AMQP
// request prior to it being decoded. This is useful for obtaining anything off
// of the response and adding onto the context prior to decoding.
181649
func PublisherAfter(after ...PublisherResponseFunc) PublisherOption {
	return func(p *Publisher) { p.after = append(p.after, after...) }
}
650328
func (s *StanServer) processClientPublish(m *nats.Msg) {
	iopm := &ioPendingMsg{m: m}
	pm := &iopm.pm
	if pm.Unmarshal(m.Data) != nil {
		if s.processCtrlMsg(m) {
			return
		}
		// else we will report an error below...
	}

	// Make sure we have a guid and valid channel name.
	if pm.Guid == "" || !util.IsChannelNameValid(pm.Subject, false) {
		s.log.Errorf("Received invalid client publish message %v", pm)
		s.sendPublishErr(m.Reply, pm.Guid, ErrInvalidPubReq)
		return
	}

	if s.debug {
		s.log.Tracef("[Client:%s] Received message from publisher subj=%s guid=%s", pm.ClientID, pm.Subject, pm.Guid)
	}

	// Check if the client is valid. We do this after the clustered check so
	// that only the leader performs this check.
	valid := false
	if s.partitions != nil {
		// In partitioning mode it is possible that we get there
		// before the connect request is processed. If so, make sure we wait
		// for conn request	to be processed first. Check clientCheckTimeout
		// doc for details.
		valid = s.clients.isValidWithTimeout(pm.ClientID, pm.ConnID, clientCheckTimeout)
	} else {
		valid = s.clients.isValid(pm.ClientID, pm.ConnID)
	}
	if !valid {
		s.log.Errorf("Received invalid client publish message %v", pm)
		s.sendPublishErr(m.Reply, pm.Guid, ErrInvalidPubReq)
		return
	}

	s.ioChannel <- iopm
}
931271
func (g *MetagameServer) simpleAfter(ctx context.Context, resp interface{}, err error) (interface{}, error) {
	logger := pitaya.GetDefaultLoggerFromCtx(ctx)
	logger.Infof("Simple After exec - response: %v , error: %v", resp, err)

	return resp, err
}
489274
func (b *Builder) ToClientResponse() (*tap.Message, error) {
	t := tap.Message_CLIENT_RESPONSE
	return &tap.Message{
		Type:             &t,
		SocketFamily:     &b.SocketFam,
		SocketProtocol:   &b.SocketProto,
		ResponseTimeSec:  &b.TimeSec,
		ResponseTimeNsec: &b.TimeNsec,
		ResponseMessage:  b.Packed,
		QueryAddress:     b.Address,
		QueryPort:        &b.Port,
	}, b.err
}
360720
func DefaultResponsePublisher(
	ctx context.Context,
	deliv *amqp.Delivery,
	ch Channel,
	pub *amqp.Publishing,
) error {
	if pub.CorrelationId == "" {
		pub.CorrelationId = deliv.CorrelationId
	}

	replyExchange := getPublishExchange(ctx)
	replyTo := getPublishKey(ctx)
	if replyTo == "" {
		replyTo = deliv.ReplyTo
	}

	return ch.Publish(
		replyExchange,
		replyTo,
		false, // mandatory
		false, // immediate
		*pub,
	)
}
26142
func (s *SetReceiptRulePositionInput) SetAfter(v string) *SetReceiptRulePositionInput {
	s.After = &v
	return s
}
1163553
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		var user string
		audit, ok := ctx.Value("x-audit-user").(string)
		if ok && audit != "" {
			user = audit
		}

		md, ok := metadata.FromIncomingContext(ctx)
		if ok {
			if len(md["x-audit-user"]) > 0 {
				user = md["x-audit-user"][0]
			}
		}

		m.Metadata["x-audit-user"] = user
		return next(ctx, topic, m)
	}
}
610669
func publish(sessions chan chan session, messages <-chan message) {
	for session := range sessions {
		var (
			running bool
			reading = messages
			pending = make(chan message, 1)
			confirm = make(chan amqp.Confirmation, 1)
		)

		pub := <-session

		// publisher confirms for this channel/connection
		if err := pub.Confirm(false); err != nil {
			log.Printf("publisher confirms not supported")
			close(confirm) // confirms not supported, simulate by always nacking
		} else {
			pub.NotifyPublish(confirm)
		}

		log.Printf("publishing...")

	Publish:
		for {
			var body message
			select {
			case confirmed, ok := <-confirm:
				if !ok {
					break Publish
				}
				if !confirmed.Ack {
					log.Printf("nack message %d, body: %q", confirmed.DeliveryTag, string(body))
				}
				reading = messages

			case body = <-pending:
				routingKey := "ignored for fanout exchanges, application dependent for other exchanges"
				err := pub.Publish(exchange, routingKey, false, false, amqp.Publishing{
					Body: body,
				})
				// Retry failed delivery on the next session
				if err != nil {
					pending <- body
					pub.Close()
					break Publish
				}

			case body, running = <-reading:
				// all messages consumed
				if !running {
					return
				}
				// work on pending delivery until ack'd
				pending <- body
				reading = nil
			}
		}
	}
}
635169
func NewSubscriber(
	e endpoint.Endpoint,
	dec DecodeRequestFunc,
	enc EncodeResponseFunc,
	options ...SubscriberOption,
) *Subscriber {
	s := &Subscriber{
		e:                 e,
		dec:               dec,
		enc:               enc,
		responsePublisher: DefaultResponsePublisher,
		errorEncoder:      DefaultErrorEncoder,
		errorHandler:      transport.NewLogErrorHandler(log.NewNopLogger()),
	}
	for _, option := range options {
		option(s)
	}
	return s
}
1258069
func (l *PacketList) InsertAfter(v Packet, mark *PacketElement) *PacketElement {
	if mark.list != l {
		return nil
	}
	// see comment in List.Remove about initialization of l
	return l.insertValue(v, mark)
}
440351
func (n *Stan) PublishAsync(subject string, data PayloadWithContext, ackH stan.AckHandler) (string, error) {
	n.updateContext(data, RequestType_PUB)
	bData, err := proto.Marshal(data)
	if err != nil {
		return "", err
	}
	return n.Con.PublishAsync(subject, bData, ackH)
}
845565
func (p *ExamplePlugin) etcdPublisher() {
	// Wait for the consumer to initialize
	time.Sleep(1 * time.Second)

	p.Log.Print("KeyValPublisher started")

	// Convert data into the proto format.
	exampleData := p.buildData("string1", 0, true)

	// PUT: demonstrate how to use the Data Broker Put() API to store
	// a simple data structure into ETCD.
	label := etcdKeyPrefixLabel(p.ServiceLabel.GetAgentLabel(), "index")
	p.Log.Infof("Write data to %v", label)
	if err := p.Publisher.Put(label, exampleData); err != nil {
		p.Log.Fatal(err)
	}

	// Prepare different set of data.
	p.Log.Infof("Update data at %v", label)
	exampleData = p.buildData("string2", 1, false)

	// UPDATE: demonstrate how use the Data Broker Put() API to change
	// an already stored data in ETCD.
	if err := p.Publisher.Put(label, exampleData); err != nil {
		p.Log.Fatal(err)
	}

	// Prepare another different set of data.
	p.Log.Infof("Update data at %v", label)
	exampleData = p.buildData("string3", 2, false)

	// UPDATE: only to demonstrate Unregister functionality
	if err := p.Publisher.Put(label, exampleData); err != nil {
		p.Log.Fatal(err)
	}

	// Wait for the consumer (change should not be passed to listener)
	time.Sleep(2 * time.Second)
	p.publisherDone = true
}
1296132
func (w *clientWrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) (err error) {
	t := newPublicationTracker(p, ClientProfile)
	ctx = t.start(ctx, true)

	defer func() { t.end(ctx, err) }()

	ctx = injectTraceIntoCtx(ctx, t.span)

	err = w.Client.Publish(ctx, p, opts...)
	return
}
44537
func (pb *ProgressBar) GetPrintAfter() string {
	pb.lock.Lock()
	val := pb.printAfter
	pb.lock.Unlock()
	return val
}
693633
func NewPublisher(
	ch Channel,
	q *amqp.Queue,
	enc EncodeRequestFunc,
	dec DecodeResponseFunc,
	options ...PublisherOption,
) *Publisher {
	p := &Publisher{
		ch:        ch,
		q:         q,
		enc:       enc,
		dec:       dec,
		deliverer: DefaultDeliverer,
		timeout:   10 * time.Second,
	}
	for _, option := range options {
		option(p)
	}
	return p
}
503701
func ClientBefore(before ...httptransport.RequestFunc) ClientOption {
	return func(c *Client) { c.before = append(c.before, before...) }
}
1195646
func (t *txSubscriptionClient) notificationProxier() {
out:
	for {
		select {
		case txNtfn := <-t.txClient.C:
			// TODO(roasbeef): handle detached blocks
			currentHeight := t.w.Manager.SyncedTo().Height

			// Launch a goroutine to re-package and send
			// notifications for any newly confirmed transactions.
			go func() {
				for _, block := range txNtfn.AttachedBlocks {
					details, err := minedTransactionsToDetails(currentHeight, block, t.w.ChainParams())
					if err != nil {
						continue
					}

					for _, d := range details {
						select {
						case t.confirmed <- d:
						case <-t.quit:
							return
						}
					}
				}

			}()

			// Launch a goroutine to re-package and send
			// notifications for any newly unconfirmed transactions.
			go func() {
				for _, tx := range txNtfn.UnminedTransactions {
					detail, err := unminedTransactionsToDetail(tx)
					if err != nil {
						continue
					}

					select {
					case t.unconfirmed <- detail:
					case <-t.quit:
						return
					}
				}
			}()
		case <-t.quit:
			break out
		}
	}

	t.wg.Done()
}