// PublishAfter publishes to ALL subscribers. Usefull in client applications
// who reuse the publish connection for subscribing
434257
func PublishAfter(tags []string, data string, delay time.Duration) error {
	go func() {
		<-time.After(delay)
		if err := Publish(tags, data); err != nil {
			// log this error and continue?
			lumber.Error("Failed to PublishAfter - %s", err.Error())
		}
	}()

	return nil
}
666258
func ClientAfter(after ...ClientResponseFunc) ClientOption {
	return func(c *Client) { c.after = append(c.after, after...) }
}
1255043
func (s *Service) PubSubSubscriberClient(c context.Context, projectID string) (*pubsub.Client, error) {
	ts, err := serverAuth.GetTokenSource(
		c, serverAuth.AsSelf,
		serverAuth.WithScopes(gcps.SubscriberScopes...))
	if err != nil {
		return nil, err
	}
	return pubsub.NewClient(c, projectID,
		option.WithUserAgent(s.getUserAgent()),
		option.WithTokenSource(ts))
}
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
}
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)
}
1257701
func (w *WebHookPayload) GetAfter() string {
	if w == nil || w.After == nil {
		return ""
	}
	return *w.After
}
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
}
1052002
func (o Middleware) PublisherMsgInterceptor(serviceName string, next pubsub.PublishHandler) pubsub.PublishHandler {
	return func(ctx context.Context, topic string, m *pubsub.Msg) error {
		start := time.Now()
		err := next(ctx, topic, m)

		publishedCounter.
			WithLabelValues(topic, serviceName).Inc()
		publishDurationsHistogram.
			Observe(time.Now().Sub(start).Seconds())
		publishedSize.
			WithLabelValues(topic, serviceName).
			Add(float64(len(m.Data)))

		return err
	}
}
1133801
func (t *Module) SetAfter(action cli.ActionFunc) *Module {
	t.After = action
	return t
}
1276513
func (r *SearchRequest) SearchAfter(sortValues ...interface{}) *SearchRequest {
	r.searchSource = r.searchSource.SearchAfter(sortValues...)
	return r
}
411866
func (nc *Conn) PublishRequest(subj, reply string, data []byte) error {
	return nc.publish(subj, reply, data)
}
752913
func (s *DominantLanguageDetectionJobFilter) SetSubmitTimeAfter(v time.Time) *DominantLanguageDetectionJobFilter {
	s.SubmitTimeAfter = &v
	return s
}
105730
func (p *PushEvent) GetAfter() string {
	if p == nil || p.After == nil {
		return ""
	}
	return *p.After
}
509586
func SubscriberAfter(after ...SubscriberResponseFunc) SubscriberOption {
	return func(s *Subscriber) { s.after = append(s.after, after...) }
}
707874
func (conn *BytesConnectionStr) NewAsyncPublisher(topic string, successClb func(*client.ProducerMessage), errorClb func(err *client.ProducerError)) (BytesPublisher, error) {
	return &bytesAsyncPublisherKafka{conn, topic, successClb, errorClb}, nil
}
856939
func Publish(ctx context.Context, topic string, msg proto.Message) *PublishResult {
	pr := &PublishResult{Ready: make(chan struct{})}
	go func() {
		err := client.Publish(ctx, topic, msg, false)
		pr.Err = err
		close(pr.Ready)
	}()
	return pr
}
93413
func (p *Provider) Subscribe(opts pubsub.HandlerOptions, h pubsub.MsgHandler) {
	logrus.Infof("Subscribing to %s successWriter/name %s", opts.Topic, opts.ServiceName+"."+opts.Name)
	logSuccess := logrus.WithField("pubsub", "kafka")
	successWriter := logSuccess.Writer()
	p.Reader = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        p.Brokers,
		GroupID:        opts.ServiceName + "." + opts.Name,
		Topic:          opts.Topic,
		MaxBytes:       10e6, // 10MB
		CommitInterval: time.Second,
		MaxWait:        50 * time.Millisecond,
		Logger:         log.New(successWriter, "", log.Lshortfile),
		ErrorLogger:    log.New(ioutil.Discard, "", log.Lshortfile),
	})

	b := &backoff.Backoff{
		Min:    200 * time.Millisecond,
		Max:    600 * time.Second,
		Factor: 2,
		Jitter: true,
	}

	go func() {
		for {
			ctx := context.Background()
			m, err := p.Reader.FetchMessage(ctx)
			if err != nil {
				d := b.Duration()
				logrus.Errorf(
					"Subscription receive to topic %s failed, reconnecting in %v. Err: %v",
					opts.Topic, d, err,
				)
				time.Sleep(d)
			}

			b.Reset()

			msg := pubsub.Msg{
				ID:   string(m.Key),
				Data: m.Value,
				Ack: func() {
					p.Reader.CommitMessages(ctx, m)
				},
				Nack: func() {},
			}

			err = h(ctx, msg)
			if err != nil {
				break
			}

			if opts.AutoAck {
				msg.Ack()
			}

			logrus.Debugf("message at topic/partition/offset %v/%v/%v\n",
				m.Topic, m.Partition, m.Offset)
		}
		successWriter.Close()
	}()
}