From 17394e7e40404431f15f4ca7708a682abb314c9f Mon Sep 17 00:00:00 2001 From: Naoki Kosaka Date: Sun, 28 Jul 2019 18:43:50 +0900 Subject: [PATCH] Add redis pub/sub support. --- State/state.go | 50 +++++++++++++++++++++++++++++++++++++++------ State/state_test.go | 12 +++++------ cli/cli.go | 2 +- cli/cli_test.go | 2 ++ handle.go | 1 - main.go | 3 ++- main_test.go | 2 ++ 7 files changed, 57 insertions(+), 15 deletions(-) diff --git a/State/state.go b/State/state.go index f9209fd..942c8b9 100644 --- a/State/state.go +++ b/State/state.go @@ -1,6 +1,7 @@ package state import ( + "fmt" "strings" "github.com/go-redis/redis" @@ -19,9 +20,10 @@ const ( ) // NewState : Create new RelayState instance with redis client -func NewState(redisClient *redis.Client) RelayState { +func NewState(redisClient *redis.Client, notify bool) RelayState { var config RelayState config.RedisClient = redisClient + config.notify = notify config.Load() return config @@ -30,6 +32,7 @@ func NewState(redisClient *redis.Client) RelayState { // RelayState : Store subscriptions and relay configrations type RelayState struct { RedisClient *redis.Client + notify bool RelayConfig relayConfig `json:"relayConfig,omitempty"` LimitedDomains []string `json:"limitedDomains,omitempty"` @@ -37,6 +40,21 @@ type RelayState struct { Subscriptions []Subscription `json:"subscriptions,omitempty"` } +func (config *RelayState) ListenNotify() { + go func() { + _, err := config.RedisClient.Subscribe("relay_refresh").Receive() + if err != nil { + panic(err) + } + ch := config.RedisClient.Subscribe("relay_refresh").Channel() + + for range ch { + fmt.Println("Config refreshed from state changed notify.") + config.Load() + } + }() +} + // Load : Refrash content from redis func (config *RelayState) Load() { config.RelayConfig.load(config.RedisClient) @@ -84,7 +102,11 @@ func (config *RelayState) SetConfig(key Config, value bool) { case CreateAsAnnounce: config.RedisClient.HSet("relay:config", "create_as_announce", strValue).Result() } - config.Load() + if config.notify { + config.RedisClient.Publish("relay_refresh", "Config refreshing request.") + } else { + config.Load() + } } // AddSubscription : Add new instance for subscription list @@ -95,7 +117,11 @@ func (config *RelayState) AddSubscription(domain Subscription) { "actor_id": domain.ActorID, }) - config.Load() + if config.notify { + config.RedisClient.Publish("relay_refresh", "Config refreshing request.") + } else { + config.Load() + } } // DelSubscription : Delete instance from subscription list @@ -103,7 +129,11 @@ func (config *RelayState) DelSubscription(domain string) { config.RedisClient.Del("relay:subscription:" + domain).Result() config.RedisClient.Del("relay:pending:" + domain).Result() - config.Load() + if config.notify { + config.RedisClient.Publish("relay_refresh", "Config refreshing request.") + } else { + config.Load() + } } // SelectSubscription : Select instance from string @@ -124,7 +154,11 @@ func (config *RelayState) SetBlockedDomain(domain string, value bool) { config.RedisClient.HDel("relay:config:blockedDomain", domain).Result() } - config.Load() + if config.notify { + config.RedisClient.Publish("relay_refresh", "Config refreshing request.") + } else { + config.Load() + } } // SetLimitedDomain : Set/Unset instance for limited domain @@ -135,7 +169,11 @@ func (config *RelayState) SetLimitedDomain(domain string, value bool) { config.RedisClient.HDel("relay:config:limitedDomain", domain).Result() } - config.Load() + if config.notify { + config.RedisClient.Publish("relay_refresh", "Config refreshing request.") + } else { + config.Load() + } } // Subscription : Instance subscription information diff --git a/State/state_test.go b/State/state_test.go index 3162b38..e6f8114 100644 --- a/State/state_test.go +++ b/State/state_test.go @@ -32,7 +32,7 @@ func TestMain(m *testing.M) { func TestInitialLoad(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) if testState.RelayConfig.BlockService != false { t.Fatalf("Failed read config.") @@ -49,7 +49,7 @@ func TestInitialLoad(t *testing.T) { func TestAddLimited(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) testState.SetLimitedDomain("example.com", true) @@ -79,7 +79,7 @@ func TestAddLimited(t *testing.T) { func TestAddBlocked(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) testState.SetBlockedDomain("example.com", true) @@ -109,7 +109,7 @@ func TestAddBlocked(t *testing.T) { func TestAddSubscription(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) testState.AddSubscription(Subscription{ Domain: "example.com", @@ -142,7 +142,7 @@ func TestAddSubscription(t *testing.T) { func TestLoadCompatiSubscription(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) testState.AddSubscription(Subscription{ Domain: "example.com", @@ -167,7 +167,7 @@ func TestLoadCompatiSubscription(t *testing.T) { func TestSetConfig(t *testing.T) { redisClient.FlushAll().Result() - testState := NewState(redisClient) + testState := NewState(redisClient, false) testState.SetConfig(BlockService, true) if testState.RelayConfig.BlockService != true { diff --git a/cli/cli.go b/cli/cli.go index 9a51c55..8aeefaf 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -58,7 +58,7 @@ func initConfig() { panic(err) } redisClient := redis.NewClient(redisOption) - relayState = state.NewState(redisClient) + relayState = state.NewState(redisClient, true) var machineryConfig = &config.Config{ Broker: viper.GetString("redis_url"), DefaultQueue: "relay", diff --git a/cli/cli_test.go b/cli/cli_test.go index 14cf9b7..5993338 100644 --- a/cli/cli_test.go +++ b/cli/cli_test.go @@ -5,12 +5,14 @@ import ( "testing" "github.com/spf13/viper" + state "github.com/yukimochi/Activity-Relay/State" ) func TestMain(m *testing.M) { viper.Set("actor_pem", "../misc/testKey.pem") viper.Set("relay_domain", "relay.yukimochi.example.org") initConfig() + relayState = state.NewState(relayState.RedisClient, false) relayState.RedisClient.FlushAll().Result() code := m.Run() diff --git a/handle.go b/handle.go index 4f9925a..f5f5a0d 100644 --- a/handle.go +++ b/handle.go @@ -176,7 +176,6 @@ func handleInbox(writer http.ResponseWriter, request *http.Request, activityDeco writer.WriteHeader(400) writer.Write(nil) } else { - relayState.Load() domain, _ := url.Parse(activity.Actor) switch activity.Type { case "Follow": diff --git a/main.go b/main.go index d5ff4f1..2c12dda 100644 --- a/main.go +++ b/main.go @@ -58,7 +58,8 @@ func initConfig() { panic(err) } redisClient := redis.NewClient(redisOption) - relayState = state.NewState(redisClient) + relayState = state.NewState(redisClient, true) + relayState.ListenNotify() machineryConfig := &config.Config{ Broker: viper.GetString("redis_url"), DefaultQueue: "relay", diff --git a/main_test.go b/main_test.go index 7eef040..3895891 100644 --- a/main_test.go +++ b/main_test.go @@ -5,12 +5,14 @@ import ( "testing" "github.com/spf13/viper" + state "github.com/yukimochi/Activity-Relay/State" ) func TestMain(m *testing.M) { viper.Set("actor_pem", "misc/testKey.pem") viper.Set("relay_domain", "relay.yukimochi.example.org") initConfig() + relayState = state.NewState(relayState.RedisClient, false) // Load Config code := m.Run()