// readAndPublish reads to the delim from r and publishes the bytes
// to the map of producers.
171483
func readAndPublish(r *bufio.Reader, delim byte, producers map[string]*nsq.Producer) error {
	line, readErr := r.ReadBytes(delim)

	if len(line) > 0 {
		// trim the delimiter
		line = line[:len(line)-1]
	}

	if len(line) == 0 {
		return readErr
	}

	for _, producer := range producers {
		err := producer.Publish(*topic, line)
		if err != nil {
			return err
		}
	}

	return readErr
}
1093141
func NewPUBLISH(opts *PUBLISHOptions) (Packet, error) {
	// Initialize the options.
	if opts == nil {
		opts = &PUBLISHOptions{}
	}

	// Validate the options.
	if err := opts.validate(); err != nil {
		return nil, err
	}

	// Create a PUBLISH Packet.
	p := &PUBLISH{
		DUP:       opts.DUP,
		QoS:       opts.QoS,
		retain:    opts.Retain,
		TopicName: opts.TopicName,
		PacketID:  opts.PacketID,
		Message:   opts.Message,
	}

	// Set the variable header to the Packet.
	p.setVariableHeader()

	// Set the payload to the Packet.
	p.setPayload()

	// Set the Fixed header to the Packet.
	p.setFixedHeader()

	// Return the Packet.
	return p, nil
}
1001051
func NewPUBLISHFromBytes(fixedHeader FixedHeader, remaining []byte) (Packet, error) {
	// Validate the byte data.
	if err := validatePUBLISHBytes(fixedHeader, remaining); err != nil {
		return nil, err
	}

	// Get the first byte from the fixedHeader.
	b := fixedHeader[0]

	// Create a PUBLISH Packet.
	p := &PUBLISH{
		DUP:    b&0x08 == 0x08,
		QoS:    b & 0x06 >> 1,
		retain: b&0x01 == 0x01,
	}

	// Set the fixed header to the Packet.
	p.fixedHeader = fixedHeader

	// Extract the length of the Topic Name.
	lenTopicName, _ := decodeUint16(remaining[0:2])

	// Calculate the length of the variable header.
	var lenVariableHeader int

	if p.QoS == mqtt.QoS0 {
		lenVariableHeader = 2 + int(lenTopicName)
	} else {
		lenVariableHeader = 2 + int(lenTopicName) + 2
	}

	// Set the variable header to the Packet.
	p.variableHeader = remaining[:lenVariableHeader]

	// Set the payload to the Packet.
	p.payload = remaining[lenVariableHeader:]

	// Set the Topic Name to the Packet.
	p.TopicName = remaining[2 : 2+lenTopicName]

	// Extract the Packet Identifier.
	var packetID uint16

	if p.QoS != mqtt.QoS0 {
		packetID, _ = decodeUint16(remaining[2+lenTopicName : 2+lenTopicName+2])
	}

	// Set the Packet Identifier to the Packet.
	p.PacketID = packetID

	// Set the Application Message to the Packet.
	p.Message = p.payload

	// Return the Packet.
	return p, nil
}
359216
func (g *Gnostic) readOpenAPIBinary(data []byte) (message proto.Message, err error) {
	// try to read an OpenAPI v3 document
	documentV3 := &openapi_v3.Document{}
	err = proto.Unmarshal(data, documentV3)
	if err == nil && strings.HasPrefix(documentV3.Openapi, "3.0") {
		g.sourceFormat = SourceFormatOpenAPI3
		return documentV3, nil
	}
	// if that failed, try to read an OpenAPI v2 document
	documentV2 := &openapi_v2.Document{}
	err = proto.Unmarshal(data, documentV2)
	if err == nil && strings.HasPrefix(documentV2.Swagger, "2.0") {
		g.sourceFormat = SourceFormatOpenAPI2
		return documentV2, nil
	}
	// if that failed, try to read a Discovery Format document
	discoveryDocument := &discovery_v1.Document{}
	err = proto.Unmarshal(data, discoveryDocument)
	if err == nil { // && strings.HasPrefix(documentV2.Swagger, "2.0") {
		g.sourceFormat = SourceFormatDiscovery
		return discoveryDocument, nil
	}
	return nil, err
}
95953
func (cli *Client) Publish(opts *PublishOptions) error {
	// Lock for reading.
	cli.muConn.RLock()

	// Unlock.
	defer cli.muConn.RUnlock()

	// Check the Network Connection.
	if cli.conn == nil {
		return ErrNotYetConnected
	}

	// Initialize the options.
	if opts == nil {
		opts = &PublishOptions{}
	}

	// Create a PUBLISH Packet.
	p, err := cli.newPUBLISHPacket(opts)
	if err != nil {
		return err
	}

	// Send the Packet to the Server.
	cli.conn.send <- p

	return nil
}
74126
func (p *NatsPublisher) Publish(subject string, r io.Reader) error {
	_, err := p.server.initSubject(subject)
	p.server.queue[subject] <- r
	return err
}
439072
func NewPublisher(exchange string, key string, opts ...PublisherOpt) *Publisher {
	p := &Publisher{
		exchange: exchange,
		key:      key,
		pubChan:  make(chan publishMaybeErr),
		stop:     make(chan struct{}),
	}
	for _, o := range opts {
		o(p)
	}
	return p
}
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)
	}
}
841449
func MakePublishInfo(networks map[string]string, ports []string) map[string][]string {
	result := map[string][]string{}
	for networkName, ip := range networks {
		data := []string{}
		for _, port := range ports {
			data = append(data, fmt.Sprintf("%s:%s", ip, port))
		}
		if len(data) > 0 {
			result[networkName] = data
		}
	}
	return result
}
1058572
func NopResponsePublisher(
	ctx context.Context,
	deliv *amqp.Delivery,
	ch Channel,
	pub *amqp.Publishing,
) error {
	return nil
}
1173937
func (c *Client) processPublish(publish *packet.Publish) error {
	// handle qos 0 flow
	if publish.Message.QOS == 0 {
		// publish message
		err := c.backend.Publish(c, &publish.Message, nil)
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)

		return nil
	}

	// acquire publish token
	select {
	case <-c.publishTokens:
		// continue
	case <-time.After(c.TokenTimeout):
		return c.die(ClientError, ErrTokenTimeout)
	case <-c.tomb.Dying():
		return tomb.ErrDying
	}

	// handle qos 1 flow
	if publish.Message.QOS == 1 {
		// prepare puback
		puback := packet.NewPuback()
		puback.ID = publish.ID

		// publish message and queue puback if ack is called
		err := c.backend.Publish(c, &publish.Message, func() {
			c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)

			select {
			case c.ackQueue <- puback:
			case <-c.tomb.Dying():
			}
		})
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
	}

	// handle qos 2 flow
	if publish.Message.QOS == 2 {
		// store received publish packet in session
		err := c.session.SavePacket(session.Incoming, publish)
		if err != nil {
			return c.die(SessionError, err)
		}

		// prepare pubrec packet
		pubrec := packet.NewPubrec()
		pubrec.ID = publish.ID

		// signal qos 2 pubrec
		err = c.send(pubrec, true)
		if err != nil {
			return c.die(TransportError, err)
		}
	}

	return nil
}
52807
func IsPublishEvent(model interface{}) (ok bool) {
	if model != nil {
		_, ok = reflect.New(utils.ModelType(model)).Interface().(PublishEventInterface)
	}
	return
}
1022240
func (r *Reader) ReadAndParseBox(typ BoxType) (Box, error) {
	box, err := r.ReadBox()
	if err != nil {
		return nil, fmt.Errorf("error reading %q box: %v", typ, err)
	}
	if box.Type() != typ {
		return nil, fmt.Errorf("error reading %q box: got box type %q instead", typ, box.Type())
	}
	pbox, err := box.Parse()
	if err != nil {
		return nil, fmt.Errorf("error parsing read %q box: %v", typ, err)
	}
	return pbox, nil
}
237197
func Publish(topic string, body []byte) error {
	return DefaultProducer.Publish(topic, body)
}
1321682
func (br *defaultBroker) Publish(pub Sender, msg *Publish) {
	pubID := NewID()
	evtTemplate := Event{
		Publication: pubID,
		Arguments:   msg.Arguments,
		ArgumentsKw: msg.ArgumentsKw,
		Details:     make(map[string]interface{}),
	}

	excludePublisher := true
	if exclude, ok := msg.Options["exclude_me"].(bool); ok {
		excludePublisher = exclude
	}

	for id, sub := range br.routes[msg.Topic] {
		// shallow-copy the template
		event := evtTemplate
		event.Subscription = id
		// don't send event to publisher
		if (excludePublisher && sub != pub) || !excludePublisher {
			sub.Send(&event)
		}
	}

	// only send published message if acknowledge is present and set to true
	if doPub, _ := msg.Options["acknowledge"].(bool); doPub {
		pub.Send(&Published{Request: msg.Request, Publication: pubID})
	}
}
972151
func (p DestinationPublisher) Publish(d Descriptor) error {
	if d.TargetPath == "" {
		return errors.New("Publish: must provide a TargetPath")
	}

	src := d.Src

	transformers := p.createTransformerChain(d)

	if len(transformers) != 0 {
		b := bp.GetBuffer()
		defer bp.PutBuffer(b)

		if err := transformers.Apply(b, d.Src); err != nil {
			return err
		}

		// This is now what we write to disk.
		src = b
	}

	f, err := helpers.OpenFileForWriting(p.fs, d.TargetPath)
	if err != nil {
		return err
	}
	defer f.Close()

	_, err = io.Copy(f, src)
	if err == nil && d.StatCounter != nil {
		atomic.AddUint64(d.StatCounter, uint64(1))
	}
	return err
}
1346660
func (o *offsetReader) CloseAndCopy(offset int64) *offsetReader {
	o.Close()
	return newOffsetReader(o.buf, offset)
}