-
Notifications
You must be signed in to change notification settings - Fork 84
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(store-forward): Enable Custom Factory Registration (#1051)
Move necessary contracts to public package. Required moving store client initialization from the end of service.Initialize to the beginning of service.MakeItRun so there is a chance to register custom store implementations. Fixes #1045 Signed-off-by: Alex Ullrich <[email protected]>
- Loading branch information
Showing
20 changed files
with
338 additions
and
280 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| // | ||
| // Copyright (c) 2021 One Track Consulting | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package app | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/bootstrap/container" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/common" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/store/db" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/store/db/redis" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/pkg/interfaces" | ||
| bootstrapContainer "github.com/edgexfoundry/go-mod-bootstrap/v2/bootstrap/container" | ||
| "github.com/edgexfoundry/go-mod-bootstrap/v2/bootstrap/environment" | ||
| "github.com/edgexfoundry/go-mod-bootstrap/v2/bootstrap/secret" | ||
| bootstrapConfig "github.com/edgexfoundry/go-mod-bootstrap/v2/config" | ||
| "github.com/edgexfoundry/go-mod-bootstrap/v2/di" | ||
| "strings" | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| // RegisterCustomStoreFactory allows registration of alternative storage implementation to back the Store&Forward loop | ||
| func (svc *Service) RegisterCustomStoreFactory(name string, factory func(cfg interfaces.DatabaseInfo, cred bootstrapConfig.Credentials) (interfaces.StoreClient, error)) error { | ||
| if name == db.RedisDB { | ||
| return fmt.Errorf("cannot register factory for reserved name %q", name) | ||
| } | ||
|
|
||
| if svc.customStoreClientFactories == nil { | ||
| svc.customStoreClientFactories = make(map[string]func(db interfaces.DatabaseInfo, cred bootstrapConfig.Credentials) (interfaces.StoreClient, error), 1) | ||
| } | ||
|
|
||
| svc.customStoreClientFactories[strings.ToUpper(name)] = factory | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (svc *Service) createStoreClient(database interfaces.DatabaseInfo, credentials bootstrapConfig.Credentials) (interfaces.StoreClient, error) { | ||
| switch strings.ToLower(database.Type) { | ||
| case db.RedisDB: | ||
| return redis.NewClient(database, credentials) | ||
| default: | ||
| if factory, found := svc.customStoreClientFactories[strings.ToUpper(database.Type)]; found { | ||
| return factory(database, credentials) | ||
| } | ||
| return nil, db.ErrUnsupportedDatabase | ||
| } | ||
| } | ||
|
|
||
| func (svc *Service) startStoreForward() { | ||
| var storeForwardEnabledCtx context.Context | ||
| svc.ctx.storeForwardWg = &sync.WaitGroup{} | ||
| storeForwardEnabledCtx, svc.ctx.storeForwardCancelCtx = context.WithCancel(context.Background()) | ||
| svc.runtime.StartStoreAndForward(svc.ctx.appWg, svc.ctx.appCtx, svc.ctx.storeForwardWg, storeForwardEnabledCtx, svc.serviceKey) | ||
| } | ||
|
|
||
| func (svc *Service) stopStoreForward() { | ||
| svc.LoggingClient().Info("Canceling Store and Forward retry loop") | ||
| svc.ctx.storeForwardCancelCtx() | ||
| svc.ctx.storeForwardWg.Wait() | ||
| } | ||
|
|
||
| func initializeStoreClient(config *common.ConfigurationStruct, svc *Service) error { | ||
| // Only need the database client if Store and Forward is enabled | ||
| if !config.Writable.StoreAndForward.Enabled { | ||
| svc.dic.Update(di.ServiceConstructorMap{ | ||
| container.StoreClientName: func(get di.Get) interface{} { | ||
| return nil | ||
| }, | ||
| }) | ||
| return nil | ||
| } | ||
|
|
||
| logger := bootstrapContainer.LoggingClientFrom(svc.dic.Get) | ||
| secretProvider := bootstrapContainer.SecretProviderFrom(svc.dic.Get) | ||
|
|
||
| var err error | ||
|
|
||
| secrets, err := secretProvider.GetSecret(config.Database.Type) | ||
| if err != nil { | ||
| return fmt.Errorf("unable to get Database Credentials for Store and Forward: %s", err.Error()) | ||
| } | ||
|
|
||
| credentials := bootstrapConfig.Credentials{ | ||
| Username: secrets[secret.UsernameKey], | ||
| Password: secrets[secret.PasswordKey], | ||
| } | ||
|
|
||
| startup := environment.GetStartupInfo(svc.serviceKey) | ||
|
|
||
| tryUntil := time.Now().Add(time.Duration(startup.Duration) * time.Second) | ||
|
|
||
| var storeClient interfaces.StoreClient | ||
| for time.Now().Before(tryUntil) { | ||
| if storeClient, err = svc.createStoreClient(config.Database, credentials); err != nil { | ||
| logger.Warnf("unable to initialize Database '%s' for Store and Forward: %s", config.Database.Type, err.Error()) | ||
| time.Sleep(time.Duration(startup.Interval) * time.Second) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| if err != nil { | ||
| return fmt.Errorf("initialize Database for Store and Forward failed: %s", err.Error()) | ||
| } | ||
|
|
||
| svc.dic.Update(di.ServiceConstructorMap{ | ||
| container.StoreClientName: func(get di.Get) interface{} { | ||
| return storeClient | ||
| }, | ||
| }) | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| // | ||
| // Copyright (c) 2022 One Track Consulting | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package app | ||
|
|
||
| import ( | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/store/db" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/internal/store/db/redis" | ||
| "github.com/edgexfoundry/app-functions-sdk-go/v2/pkg/interfaces" | ||
| "github.com/edgexfoundry/go-mod-bootstrap/v2/config" | ||
| "github.com/google/uuid" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "strings" | ||
| "testing" | ||
| ) | ||
|
|
||
| func Test_service_RegisterCustomStoreFactory_Inbuilt_Name(t *testing.T) { | ||
| sut := &Service{} | ||
|
|
||
| err := sut.RegisterCustomStoreFactory(db.RedisDB, nil) | ||
|
|
||
| require.Error(t, err) | ||
| } | ||
|
|
||
| func Test_service_RegisterCustomStoreFactory(t *testing.T) { | ||
| sut := &Service{} | ||
|
|
||
| f := func(info interfaces.DatabaseInfo, credentials config.Credentials) (interfaces.StoreClient, error) { | ||
| return nil, nil | ||
| } | ||
|
|
||
| name := uuid.NewString() | ||
|
|
||
| err := sut.RegisterCustomStoreFactory(name, f) | ||
|
|
||
| assert.NoError(t, err) | ||
|
|
||
| _, found := sut.customStoreClientFactories[strings.ToUpper(name)] | ||
| assert.True(t, found) | ||
| } | ||
|
|
||
| func Test_service_NewStoreClient_Invalid(t *testing.T) { | ||
| sut := &Service{} | ||
|
|
||
| _, err := sut.createStoreClient(interfaces.DatabaseInfo{}, config.Credentials{}) | ||
|
|
||
| assert.Error(t, err) | ||
| } | ||
|
|
||
| func Test_service_NewStoreClient_Redis(t *testing.T) { | ||
| sut := &Service{customStoreClientFactories: map[string]func(db interfaces.DatabaseInfo, cred config.Credentials) (interfaces.StoreClient, error){}} | ||
|
|
||
| result, err := sut.createStoreClient(interfaces.DatabaseInfo{Type: db.RedisDB, Timeout: "5s"}, config.Credentials{}) | ||
|
|
||
| assert.NoError(t, err) | ||
| c, ok := result.(*redis.Client) | ||
| assert.NotNil(t, c) | ||
| assert.True(t, ok) | ||
| } | ||
|
|
||
| func Test_service_NewStoreClient_Custom(t *testing.T) { | ||
| var x interfaces.StoreClient | ||
|
|
||
| sut := &Service{} | ||
|
|
||
| f := func(info interfaces.DatabaseInfo, credentials config.Credentials) (interfaces.StoreClient, error) { | ||
| return x, nil | ||
| } | ||
|
|
||
| name := uuid.NewString() | ||
|
|
||
| err := sut.RegisterCustomStoreFactory(name, f) | ||
|
|
||
| require.NoError(t, err) | ||
|
|
||
| result, err := sut.createStoreClient(interfaces.DatabaseInfo{Type: name}, config.Credentials{}) | ||
|
|
||
| assert.NoError(t, err) | ||
| assert.Equal(t, x, result) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.