small changes and renaming

core/ctrl.go

@@ -33,7 +33,7 @@ type Controller struct {
 	mdnsSvc mdns.Service
 	pubsub  *pubsub.PubSub
 
-	manager   pubsubManager
+	psManager pubsubManager
 	denylist  pubsub.Blacklist
 	subFilter pubsub.SubscriptionFilter
 

core/pubsub.go

@@ -17,7 +17,7 @@ import (
 )
 
 var (
-	inspectInterval = time.Minute
+	peerScoreInspectionInterval = time.Minute
 )
 
 func (c *Controller) setupPubsubRouter(ctx context.Context, cfg commons.Config) error {
@@ -34,7 +34,7 @@ func (c *Controller) setupPubsubRouter(ctx context.Context, cfg commons.Config)
 
 	if cfg.Pubsub.Scoring != nil {
 		opts = append(opts, pubsub.WithPeerScore(gossip.PeerScores(*cfg.Pubsub)))
-		opts = append(opts, pubsub.WithPeerScoreInspect(c.inspectPeerScores, inspectInterval))
+		opts = append(opts, pubsub.WithPeerScoreInspect(c.inspectPeerScores, peerScoreInspectionInterval))
 	}
 
 	if cfg.Pubsub.MaxMessageSize > 0 {
@@ -48,8 +48,6 @@ func (c *Controller) setupPubsubRouter(ctx context.Context, cfg commons.Config)
 	denylist := pubsub.NewMapBlacklist()
 	opts = append(opts, pubsub.WithBlacklist(denylist))
 
-	// pubsub.WithDefaultValidator() // TODO: check
-
 	if cfg.Pubsub.SubFilter != nil {
 		re, err := regexp.Compile(cfg.Pubsub.SubFilter.Pattern)
 		if err != nil {
@@ -73,7 +71,7 @@ func (c *Controller) setupPubsubRouter(ctx context.Context, cfg commons.Config)
 	}
 	c.pubsub = ps
 	c.denylist = denylist
-	c.manager.topics = make(map[string]*topicWrapper)
+	c.psManager.topics = make(map[string]*topicWrapper)
 
 	return nil
 }
@@ -88,7 +86,7 @@ func (c *Controller) Publish(ctx context.Context, topicName string, data []byte)
 }
 
 func (c *Controller) Leave(topicName string) error {
-	tw := c.manager.getTopicWrapper(topicName)
+	tw := c.psManager.getTopicWrapper(topicName)
 	state := tw.state.Load()
 	switch state {
 	case topicStateJoined, topicStateErr:
@@ -104,7 +102,7 @@ func (c *Controller) Leave(topicName string) error {
 }
 
 func (c *Controller) Unsubscribe(topicName string) error {
-	tw := c.manager.getTopicWrapper(topicName)
+	tw := c.psManager.getTopicWrapper(topicName)
 	if tw.state.Load() == topicStateUnknown {
 		return nil // TODO: topic not found?
 	}
@@ -162,6 +160,14 @@ func (c *Controller) listenSubscription(ctx context.Context, sub *pubsub.Subscri
 }
 
 func (c *Controller) tryJoin(topicName string) (*pubsub.Topic, error) {
+	topicW := c.psManager.getTopicWrapper(topicName)
+	if topicW != nil {
+		if topicW.state.Load() == topicStateJoining {
+			return nil, fmt.Errorf("already tring to join topic %s", topicName)
+		}
+		return topicW.topic, nil
+	}
+	c.psManager.joiningTopic(topicName)
 	opts := []pubsub.TopicOpt{}
 	cfg, ok := c.cfg.Pubsub.GetTopicConfig(topicName)
 	if ok {
@@ -170,19 +176,11 @@ func (c *Controller) tryJoin(topicName string) (*pubsub.Topic, error) {
 			opts = append(opts, pubsub.WithTopicMessageIdFn(msgID))
 		}
 	}
-	topicW := c.manager.getTopicWrapper(topicName)
-	if topicW != nil {
-		if topicW.state.Load() == topicStateJoining {
-			return nil, fmt.Errorf("already tring to join topic %s", topicName)
-		}
-		return topicW.topic, nil
-	}
-	c.manager.joiningTopic(topicName)
 	topic, err := c.pubsub.Join(topicName, opts...)
 	if err != nil {
 		return nil, err
 	}
-	c.manager.upgradeTopic(topicName, topic)
+	c.psManager.upgradeTopic(topicName, topic)
 
 	if cfg.MsgValidator != nil || c.cfg.Pubsub.MsgValidator != nil {
 		msgValConfig := commons.MsgValidationConfig{}.Defaults(c.cfg.Pubsub.MsgValidator)
@@ -204,7 +202,7 @@ func (c *Controller) tryJoin(topicName string) (*pubsub.Topic, error) {
 
 func (c *Controller) trySubscribe(topic *pubsub.Topic) (sub *pubsub.Subscription, err error) {
 	topicName := topic.String()
-	sub = c.manager.getSub(topicName)
+	sub = c.psManager.getSub(topicName)
 	if sub != nil {
 		return nil, nil
 	}
@@ -219,7 +217,7 @@ func (c *Controller) trySubscribe(topic *pubsub.Topic) (sub *pubsub.Subscription
 	if err != nil {
 		return nil, err
 	}
-	c.manager.addSub(topicName, sub)
+	c.psManager.addSub(topicName, sub)
 	return sub, nil
 }