mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
This PR is intended to merge 4 months of work in the `nexus` feature branch into `main`. The functionality it brings is: - Dispatching Nexus Tasks by namespace and task queue - Internal implementation of the Incoming Service Registry - not yet exposed - Attaching workflow close callbacks on `StartWorkflowExecutionRequest` and processing of those callbacks There's more to come and some of the callback code will be refactored into a new plugin architecture (https://github.com/temporalio/temporal/pull/5446) soon. --------- Co-authored-by: PJ Doerner <[email protected]>
102 lines
4.0 KiB
Go
102 lines
4.0 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// Copyright (c) 2020 Uber Technologies, Inc.
|
|
//
|
|
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
// of this software and associated documentation files (the "Software"), to deal
|
|
// in the Software without restriction, including without limitation the rights
|
|
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
// copies of the Software, and to permit persons to whom the Software is
|
|
// furnished to do so, subject to the following conditions:
|
|
//
|
|
// The above copyright notice and this permission notice shall be included in
|
|
// all copies or substantial portions of the Software.
|
|
//
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
// THE SOFTWARE.
|
|
|
|
package client
|
|
|
|
import (
|
|
"go.temporal.io/server/common/config"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/metrics"
|
|
p "go.temporal.io/server/common/persistence"
|
|
"go.temporal.io/server/common/persistence/cassandra"
|
|
"go.temporal.io/server/common/persistence/sql"
|
|
"go.temporal.io/server/common/resolver"
|
|
)
|
|
|
|
type (
|
|
// DataStoreFactory is a low level interface to be implemented by a datastore
|
|
// Examples of datastores are cassandra, mysql etc
|
|
DataStoreFactory interface {
|
|
// Close closes the factory
|
|
Close()
|
|
// NewTaskStore returns a new task store
|
|
NewTaskStore() (p.TaskStore, error)
|
|
// NewShardStore returns a new shard store
|
|
NewShardStore() (p.ShardStore, error)
|
|
// NewMetadataStore returns a new metadata store
|
|
NewMetadataStore() (p.MetadataStore, error)
|
|
// NewExecutionStore returns a new execution store
|
|
NewExecutionStore() (p.ExecutionStore, error)
|
|
NewQueue(queueType p.QueueType) (p.Queue, error)
|
|
NewQueueV2() (p.QueueV2, error)
|
|
// NewClusterMetadataStore returns a new metadata store
|
|
NewClusterMetadataStore() (p.ClusterMetadataStore, error)
|
|
// NewNexusIncomingServiceStore returns a new nexus service store
|
|
NewNexusIncomingServiceStore() (p.NexusIncomingServiceStore, error)
|
|
}
|
|
|
|
// AbstractDataStoreFactory creates a DataStoreFactory, can be used to implement custom datastore support outside
|
|
// of the Temporal core.
|
|
AbstractDataStoreFactory interface {
|
|
NewFactory(
|
|
cfg config.CustomDatastoreConfig,
|
|
r resolver.ServiceResolver,
|
|
clusterName string,
|
|
logger log.Logger,
|
|
metricsHandler metrics.Handler,
|
|
) DataStoreFactory
|
|
}
|
|
)
|
|
|
|
func DataStoreFactoryProvider(
|
|
clusterName ClusterName,
|
|
r resolver.ServiceResolver,
|
|
config *config.Persistence,
|
|
abstractDataStoreFactory AbstractDataStoreFactory,
|
|
logger log.Logger,
|
|
metricsHandler metrics.Handler,
|
|
) (DataStoreFactory, *FaultInjectionDataStoreFactory) {
|
|
|
|
var dataStoreFactory DataStoreFactory
|
|
defaultCfg := config.DataStores[config.DefaultStore]
|
|
switch {
|
|
case defaultCfg.Cassandra != nil:
|
|
dataStoreFactory = cassandra.NewFactory(*defaultCfg.Cassandra, r, string(clusterName), logger, metricsHandler)
|
|
case defaultCfg.SQL != nil:
|
|
dataStoreFactory = sql.NewFactory(*defaultCfg.SQL, r, string(clusterName), logger)
|
|
case defaultCfg.CustomDataStoreConfig != nil:
|
|
dataStoreFactory = abstractDataStoreFactory.NewFactory(*defaultCfg.CustomDataStoreConfig, r, string(clusterName), logger, metricsHandler)
|
|
default:
|
|
logger.Fatal("invalid config: one of cassandra or sql params must be specified for default data store")
|
|
}
|
|
|
|
var faultInjection *FaultInjectionDataStoreFactory
|
|
if defaultCfg.FaultInjection != nil {
|
|
faultInjection = NewFaultInjectionDatastoreFactory(defaultCfg.FaultInjection, dataStoreFactory)
|
|
dataStoreFactory = faultInjection
|
|
}
|
|
|
|
return dataStoreFactory, faultInjection
|
|
}
|