func (c *_Conn) subscribe(subMsg utp.Subscribe, topic *security.Topic, sub *utp.Subscription) (err error) {
	c.Lock()
	defer c.Unlock()

	key := string(topic.Key)
	if key == "" {
		key, err = security.GenerateKey(c.clientID.Contract(), topic.Topic, security.AllowNone)
		if err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
			return err
		}
	}
	if exists := c.subs.Exist(key); exists && !subMsg.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
		// The contract is handled by a remote node. Forward message to it.
		if err := Globals.Cluster.routeToContract(&subMsg, topic, message.SUBSCRIBE, &message.Message{}, c); err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe").Int64("connid", int64(c.connID)).Msg("unable to subscribe to remote topic")
			return err
		}
	} else {
		messageId, err := store.Subscription.NewID()
		if err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
			return err
		}
		if first := c.subs.Increment(topic.Topic[:topic.Size], key, messageId); first {
			// Subscribe the subscriber
			payload := make([]byte, 9)
			payload[0] = uint8(sub.DeliveryMode)
			binary.LittleEndian.PutUint32(payload[1:5], uint32(c.connID))
			binary.LittleEndian.PutUint32(payload[5:9], uint32(sub.Delay))
			if err = store.Subscription.Put(c.clientID.Contract(), messageId, topic.Topic, payload); err != nil {
				log.ErrLogger.Err(err).Str("context", "conn.subscribe").Str("topic", string(topic.Topic[:topic.Size])).Int64("connid", int64(c.connID)).Msg("unable to subscribe to topic") // Unable to subscribe
				return err
			}
			// Increment the subscription counter
			c.service.meter.Subscriptions.Inc(1)
		}
	}
	return nil
}
// subscribe subscribes to a particular topic.
subscribe subscribes to a particular topic.
// subscribe subscribes to a particular topic .
func (c *_Conn) subscribe(subMsg utp.Subscribe, topic *security.Topic, sub *utp.Subscription) (err error) {
	c.Lock()
	defer c.Unlock()
	key := string(topic.Key)
	if key == "" {
		key, err = security.GenerateKey(c.clientID.Contract(), topic.Topic, security.AllowNone)
		if err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
			return err
		}
	}
	if exists := c.subs.Exist(key); exists && !subMsg.IsForwarded && Globals.Cluster.isRemoteContract(fmt.Sprint(c.clientID.Contract())) {
		if err := Globals.Cluster.routeToContract(&subMsg, topic, message.SUBSCRIBE, &message.Message{}, c); err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe").Int64("connid", int64(c.connID)).Msg("unable to subscribe to remote topic")
			return err
		}
	} else {
		messageId, err := store.Subscription.NewID()
		if err != nil {
			log.ErrLogger.Err(err).Str("context", "conn.subscribe")
			return err
		}
		if first := c.subs.Increment(topic.Topic[:topic.Size], key, messageId); first {
			payload := make([]byte, 9)
			payload[0] = uint8(sub.DeliveryMode)
			binary.LittleEndian.PutUint32(payload[1:5], uint32(c.connID))
			binary.LittleEndian.PutUint32(payload[5:9], uint32(sub.Delay))
			if err = store.Subscription.Put(c.clientID.Contract(), messageId, topic.Topic, payload); err != nil {
				log.ErrLogger.Err(err).Str("context", "conn.subscribe").Str("topic", string(topic.Topic[:topic.Size])).Int64("connid", int64(c.connID)).Msg("unable to subscribe to topic") 
				return err
			}
			c.service.meter.Subscriptions.Inc(1)
		}
	}
	return nil
}
func ( c * _Conn ) subscribe ( subMsg utp . Subscribe , topic * security . Topic , sub * utp . Subscription ) ( err error ) { c . Lock ( ) defer c . Unlock ( ) key := string ( topic . Key ) if key == " " { key , err = security . GenerateKey ( c . clientID . Contract ( ) , topic . Topic , security . AllowNone ) if err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) return err } } if exists := c . subs . Exist ( key ) ; exists && ! subMsg . IsForwarded && Globals . Cluster . isRemoteContract ( fmt . Sprint ( c . clientID . Contract ( ) ) ) { if err := Globals . Cluster . routeToContract ( & subMsg , topic , message . SUBSCRIBE , & message . Message { } , c ) ; err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) . Int64 ( " " , int64 ( c . connID ) ) . Msg ( " " ) return err } } else { messageId , err := store . Subscription . NewID ( ) if err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) return err } if first := c . subs . Increment ( topic . Topic [ : topic . Size ] , key , messageId ) ; first { payload := make ( [ ] byte , 9 ) payload [ 0 ] = uint8 ( sub . DeliveryMode ) binary . LittleEndian . PutUint32 ( payload [ 1 : 5 ] , uint32 ( c . connID ) ) binary . LittleEndian . PutUint32 ( payload [ 5 : 9 ] , uint32 ( sub . Delay ) ) if err = store . Subscription . Put ( c . clientID . Contract ( ) , messageId , topic . Topic , payload ) ; err != nil { log . ErrLogger . Err ( err ) . Str ( " " , " " ) . Str ( " " , string ( topic . Topic [ : topic . Size ] ) ) . Int64 ( " " , int64 ( c . connID ) ) . Msg ( " " ) return err } c . service . meter . Subscriptions . Inc ( 1 ) } } return nil }
subscribe subscribes to a particular topic.
subscribe subscribes to a particular topic .
// The contract is handled by a remote node. Forward message to it. // Subscribe the subscriber // Unable to subscribe // Increment the subscription counter