71
testHooks testhooks.TestHooks,
72
dlqWriter DLQWriter,
74
>
historyFetcher := eventhandler.NewHistoryPaginatedFetcher(shardContext.GetNamespaceRegistry(), clientBean, eventSerializer, shardContext.GetLogger())
75
>
return &taskProcessorManagerImpl{
76
>
config: config,
77
>
deleteMgr: workflowDeleteManager,
78
>
engine: engine,
79
>
eventSerializer: eventSerializer,
80
>
shard: shardContext,
81
>
status: common.DaemonStatusInitialized,
82
>
replicationTaskFetcherFactory: replicationTaskFetcherFactory,
83
>
workflowCache: workflowCache,
84
>
removeHistoryFetcher: historyFetcher,
85
>
logger: shardContext.GetLogger(),
86
>
metricsHandler: shardContext.GetMetricsHandler(),
87
>
testHooks: testHooks,
88
>
dlqWriter: dlqWriter,
89
>
90
>
enableFetcher: !config.EnableReplicationStream(),
91
>
taskProcessors: make(map[string][]TaskProcessor),
92
>
taskExecutorProvider: taskExecutorProvider,
93
>
taskPollerManager: newPollerManager(shardContext.GetShardID(), shardContext.GetClusterMetadata()),
94
>
minTxAckedTaskID: persistence.EmptyQueueMessageID,
95
>
shutdownChan: make(chan struct{}),
96
>
}
97
>
}
98
99
func (r *taskProcessorManagerImpl) Start() {