func (c *_Conn) onPublish(pub utp.Publish) *types.Error {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onPublish").Int64("duration", time.Since(start).Nanoseconds()).Msg("")

	for _, pubMsg := range pub.Messages {
		//Parse the key
		topic := security.ParseKey(pubMsg.Topic)
		if topic.TopicType == security.TopicInvalid {
			return types.ErrBadRequest
		}

		// Check whether the key is 'unitdb' which means it's an API request
		if len(topic.Key) == 6 && string(topic.Key) == "unitdb" {
			c.onSpecialRequest(topic, pubMsg.Payload)
			return nil
		}

		if !c.insecure {
			wildcard, err := c.onSecureRequest(topic)
			if err != nil {
				return err
			}
			if wildcard {
				return types.ErrForbidden
			}
		}

		err := store.Message.Put(c.clientID.Contract(), topic.Topic, pubMsg.Payload, pubMsg.Ttl)
		if err != nil {
			log.Error("conn.onPublish", "store message "+err.Error())
			return types.ErrServerError
		}
		// Iterate through all subscribers and send them the message
		go c.publish(pub, topic, pubMsg)
	}

	if pub.IsForwarded {
		return nil
	}

	// acknowledge a Message
	return c.acknowledge(pub)
}
// OnPublish is a handler for Publish events.
OnPublish is a handler for Publish events.
// OnPublish is a handler for Publish events .
func (c *_Conn) onPublish(pub utp.Publish) *types.Error {
	start := time.Now()
	defer log.ErrLogger.Debug().Str("context", "conn.onPublish").Int64("duration", time.Since(start).Nanoseconds()).Msg("")
	for _, pubMsg := range pub.Messages {
		topic := security.ParseKey(pubMsg.Topic)
		if topic.TopicType == security.TopicInvalid {
			return types.ErrBadRequest
		}
		if len(topic.Key) == 6 && string(topic.Key) == "unitdb" {
			c.onSpecialRequest(topic, pubMsg.Payload)
			return nil
		}
		if !c.insecure {
			wildcard, err := c.onSecureRequest(topic)
			if err != nil {
				return err
			}
			if wildcard {
				return types.ErrForbidden
			}
		}
		err := store.Message.Put(c.clientID.Contract(), topic.Topic, pubMsg.Payload, pubMsg.Ttl)
		if err != nil {
			log.Error("conn.onPublish", "store message "+err.Error())
			return types.ErrServerError
		}
		go c.publish(pub, topic, pubMsg)
	}
	if pub.IsForwarded {
		return nil
	}
	return c.acknowledge(pub)
}
func ( c * _Conn ) onPublish ( pub utp . Publish ) * types . Error { start := time . Now ( ) defer log . ErrLogger . Debug ( ) . Str ( " " , " " ) . Int64 ( " " , time . Since ( start ) . Nanoseconds ( ) ) . Msg ( " " ) for _ , pubMsg := range pub . Messages { topic := security . ParseKey ( pubMsg . Topic ) if topic . TopicType == security . TopicInvalid { return types . ErrBadRequest } if len ( topic . Key ) == 6 && string ( topic . Key ) == " " { c . onSpecialRequest ( topic , pubMsg . Payload ) return nil } if ! c . insecure { wildcard , err := c . onSecureRequest ( topic ) if err != nil { return err } if wildcard { return types . ErrForbidden } } err := store . Message . Put ( c . clientID . Contract ( ) , topic . Topic , pubMsg . Payload , pubMsg . Ttl ) if err != nil { log . Error ( " " , " " + err . Error ( ) ) return types . ErrServerError } go c . publish ( pub , topic , pubMsg ) } if pub . IsForwarded { return nil } return c . acknowledge ( pub ) }
OnPublish is a handler for Publish events.
OnPublish is a handler for Publish events .
//Parse the key // Check whether the key is 'unitdb' which means it's an API request // Iterate through all subscribers and send them the message // acknowledge a Message