func (c *_Conn) publish(pub utp.Publish, topic *security.Topic, pubMsg *utp.PublishMessage) (err error) {
	c.service.meter.InMsgs.Inc(1)
	c.service.meter.InBytes.Inc(int64(len(pubMsg.Payload)))
	// subscription count
	msgCount := 0

	subscriptions, err := store.Subscription.Get(c.clientID.Contract(), topic.Topic)
	if err != nil {
		log.ErrLogger.Err(err).Str("context", "conn.publish")
		return err
	}
	msg := &message.Message{
		MessageID: pub.MessageID,
		Topic:     string(topic.Topic[:topic.Size]),
		Payload:   pubMsg.Payload,
	}
	for _, subscription := range subscriptions {
		msg.DeliveryMode = subscription[0]
		connID := uid.LID(binary.LittleEndian.Uint32(subscription[1:5]))
		msg.Delay = int32(uid.LID(binary.LittleEndian.Uint32(subscription[5:9])))
		sub := Globals.connCache.get(connID)
		if sub != nil {
			if msg.MessageID == 0 {
				msg.MessageID = uint16(c.MessageIds.NextID(utp.PUBLISH))
			}
			switch pub.DeliveryMode {
			// Publisher's DeliveryMode RELIABLE or BATCH
			case 1, 2:
				switch msg.DeliveryMode {
				// Subscriber's DeliveryMode RELIABLE or BATCH
				case 1, 2:
					notify := &utp.ControlMessage{
						MessageType: utp.PUBLISH,
						FlowControl: utp.NOTIFY,
						MessageID:   msg.MessageID,
					}
					// persist outbound
					store.Log.PersistOutbound(uint32(sub.sessID), &pub)
					sub.send <- notify
				// Subscriber's DeliveryMode EXPRESS
				case 0:
					if !sub.SendMessage(msg) {
						log.ErrLogger.Err(err).Str("context", "conn.publish")
					}
					msgCount++
				}
			// Publisher's DeliveryMode Express
			case 0:
				if !sub.SendMessage(msg) {
					log.ErrLogger.Err(err).Str("context", "conn.publish")
				}
				msgCount++
			}
		}
	}
	c.service.meter.OutMsgs.Inc(int64(msgCount))
	c.service.meter.OutBytes.Inc(msg.Size() * int64(msgCount))

	if !pub.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
		if err = Globals.Cluster.routeToContract(&pub, topic, message.PUBLISH, msg, c); err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.publish").Int64("connid", int64(c.connID)).Msg("unable to publish to a remote topic")
			return err
		}
	}
	return nil
}
// publish publishes a message to everyone and returns the number of outgoing bytes written.
publish publishes a message to everyone and returns the number of outgoing bytes written.
// publish publishes a message to everyone and returns the number of outgoing bytes written .
func (c *_Conn) publish(pub utp.Publish, topic *security.Topic, pubMsg *utp.PublishMessage) (err error) {
	c.service.meter.InMsgs.Inc(1)
	c.service.meter.InBytes.Inc(int64(len(pubMsg.Payload)))
	msgCount := 0
	subscriptions, err := store.Subscription.Get(c.clientID.Contract(), topic.Topic)
	if err != nil {
		log.ErrLogger.Err(err).Str("context", "conn.publish")
		return err
	}
	msg := &message.Message{
		MessageID: pub.MessageID,
		Topic:     string(topic.Topic[:topic.Size]),
		Payload:   pubMsg.Payload,
	}
	for _, subscription := range subscriptions {
		msg.DeliveryMode = subscription[0]
		connID := uid.LID(binary.LittleEndian.Uint32(subscription[1:5]))
		msg.Delay = int32(uid.LID(binary.LittleEndian.Uint32(subscription[5:9])))
		sub := Globals.connCache.get(connID)
		if sub != nil {
			if msg.MessageID == 0 {
				msg.MessageID = uint16(c.MessageIds.NextID(utp.PUBLISH))
			}
			switch pub.DeliveryMode {
			case 1, 2:
				switch msg.DeliveryMode {
				case 1, 2:
					notify := &utp.ControlMessage{
						MessageType: utp.PUBLISH,
						FlowControl: utp.NOTIFY,
						MessageID:   msg.MessageID,
					}
					store.Log.PersistOutbound(uint32(sub.sessID), &pub)
					sub.send <- notify
				case 0:
					if !sub.SendMessage(msg) {
						log.ErrLogger.Err(err).Str("context", "conn.publish")
					}
					msgCount++
				}
			case 0:
				if !sub.SendMessage(msg) {
					log.ErrLogger.Err(err).Str("context", "conn.publish")
				}
				msgCount++
			}
		}
	}
	c.service.meter.OutMsgs.Inc(int64(msgCount))
	c.service.meter.OutBytes.Inc(msg.Size() * int64(msgCount))
	if !pub.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
		if err = Globals.Cluster.routeToContract(&pub, topic, message.PUBLISH, msg, c); err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.publish").Int64("connid", int64(c.connID)).Msg("unable to publish to a remote topic")
			return err
		}
	}
	return nil
}
func ( c * _Conn ) publish ( pub utp . Publish , topic * security . Topic , pubMsg * utp . PublishMessage ) ( err error ) { c . service . meter . InMsgs . Inc ( 1 ) c . service . meter . InBytes . Inc ( int64 ( len ( pubMsg . Payload ) ) ) msgCount := 0 subscriptions , err := store . Subscription . Get ( c . clientID . Contract ( ) , topic . Topic ) if err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) return err } msg := & message . Message { MessageID : pub . MessageID , Topic : string ( topic . Topic [ : topic . Size ] ) , Payload : pubMsg . Payload , } for _ , subscription := range subscriptions { msg . DeliveryMode = subscription [ 0 ] connID := uid . LID ( binary . LittleEndian . Uint32 ( subscription [ 1 : 5 ] ) ) msg . Delay = int32 ( uid . LID ( binary . LittleEndian . Uint32 ( subscription [ 5 : 9 ] ) ) ) sub := Globals . connCache . get ( connID ) if sub != nil { if msg . MessageID == 0 { msg . MessageID = uint16 ( c . MessageIds . NextID ( utp . PUBLISH ) ) } switch pub . DeliveryMode { case 1 , 2 : switch msg . DeliveryMode { case 1 , 2 : notify := & utp . ControlMessage { MessageType : utp . PUBLISH , FlowControl : utp . NOTIFY , MessageID : msg . MessageID , } store . Log . PersistOutbound ( uint32 ( sub . sessID ) , & pub ) sub . send <- notify case 0 : if ! sub . SendMessage ( msg ) { log . ErrLogger . Err ( err ) . Str ( " " , " " ) } msgCount ++ } case 0 : if ! sub . SendMessage ( msg ) { log . ErrLogger . Err ( err ) . Str ( " " , " " ) } msgCount ++ } } } c . service . meter . OutMsgs . Inc ( int64 ( msgCount ) ) c . service . meter . OutBytes . Inc ( msg . Size ( ) * int64 ( msgCount ) ) if ! pub . IsForwarded && Globals . Cluster . isRemoteContract ( fmt . Sprint ( c . clientID . Contract ( ) ) ) { if err = Globals . Cluster . routeToContract ( & pub , topic , message . PUBLISH , msg , c ) ; err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) . Int64 ( " " , int64 ( c . connID ) ) . Msg ( " " ) return err } } return nil }
publish publishes a message to everyone and returns the number of outgoing bytes written.
publish publishes a message to everyone and returns the number of outgoing bytes written .
// subscription count // Publisher's DeliveryMode RELIABLE or BATCH // Subscriber's DeliveryMode RELIABLE or BATCH // persist outbound // Subscriber's DeliveryMode EXPRESS // Publisher's DeliveryMode Express