Ronit Jain
264c3c8374
* initial commit * Follow IDQH implementation * Expect a list of data handlers from LiveNodePacket * Return null if can not subscribe * refctor to add check for subscription * initialze null * Add tests * Check subscribe retuns null/not-null * cleanup * Read all required IDQH credentails to job * Use CDQH to handle all IDQH instances * constructor abstraction to call from setjob * use flag * remove redundant because derived will call initialize on it * abstract and initialize from setjob * handle null enumerators * get creds from data handlers * handle single data handler value from data-queue-handler * Fix to support a json array * Fix missed constructor call * change access modifier to access from Tests files * Add test to get brokerageFactory from dataQueueHandler * Fix init flag to handle all conditions * Add docs * initialize from setjob * Check if websocket open before using * change defination of initialzie to include tradier * clean up * change defination * fix wrong api key name * return empty enumerator * check websocket open before sending request * check connection before subscribing * fix to include more cases * check websocket open before sending request * Minor refactoring * reafctor and use IsConnected * remove unused * clean up * Fix test cases * reverse change * include config changes * connect to websocket from setjob * check websocket connection from setjob * clean up * include condition for IDQH that are not brokerage * Address review * Add market check condition before subscribe * Remove deprecated * Minor fix for deserializing data queue handler Co-authored-by: Martin-Molinero <martin@quantconnect.com>
106 lines
4.1 KiB
C#
106 lines
4.1 KiB
C#
/*
|
|
* QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals.
|
|
* Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation.
|
|
*
|
|
* 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.
|
|
*/
|
|
|
|
using QuantConnect.Data;
|
|
using QuantConnect.Interfaces;
|
|
using QuantConnect.Packets;
|
|
using QuantConnect.Util;
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using QuantConnect.Logging;
|
|
|
|
namespace QuantConnect.Lean.Engine.DataFeeds
|
|
{
|
|
/// <summary>
|
|
/// This is an implementation of <see cref="IDataQueueHandler"/> used to handle multiple live datafeeds
|
|
/// </summary>
|
|
public class CompositeDataQueueHandler : IDataQueueHandler
|
|
{
|
|
private readonly List<IDataQueueHandler> _dataHandlers = new();
|
|
private readonly Dictionary<SubscriptionDataConfig, IDataQueueHandler> _dataConfigAndDataHandler = new();
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="CompositeDataQueueHandler"/> class
|
|
/// </summary>
|
|
public CompositeDataQueueHandler()
|
|
{
|
|
}
|
|
|
|
/// <summary>
|
|
/// Subscribe to the specified configuration
|
|
/// </summary>
|
|
/// <param name="dataConfig">defines the parameters to subscribe to a data feed</param>
|
|
/// <param name="newDataAvailableHandler">handler to be fired on new data available</param>
|
|
/// <returns>The new enumerator for this subscription request</returns>
|
|
public IEnumerator<BaseData> Subscribe(SubscriptionDataConfig dataConfig, EventHandler newDataAvailableHandler)
|
|
{
|
|
foreach (var dataHandler in _dataHandlers)
|
|
{
|
|
var enumerator = dataHandler.Subscribe(dataConfig, newDataAvailableHandler);
|
|
// Check if the enumerator is not empty
|
|
if (enumerator != null)
|
|
{
|
|
_dataConfigAndDataHandler.Add(dataConfig, dataHandler);
|
|
return enumerator;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Removes the specified configuration
|
|
/// </summary>
|
|
/// <param name="dataConfig">Subscription config to be removed</param>
|
|
public virtual void Unsubscribe(SubscriptionDataConfig dataConfig)
|
|
{
|
|
_dataConfigAndDataHandler.TryGetValue(dataConfig, out IDataQueueHandler dataHandler);
|
|
dataHandler?.Unsubscribe(dataConfig);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Sets the job we're subscribing for
|
|
/// </summary>
|
|
/// <param name="job">Job we're subscribing for</param>
|
|
public void SetJob(LiveNodePacket job)
|
|
{
|
|
var dataHandlersConfig = job.DataQueueHandler;
|
|
Log.Trace($"CompositeDataQueueHandler.SetJob(): will use {dataHandlersConfig}");
|
|
foreach (var dataHandlerName in dataHandlersConfig.DeserializeList())
|
|
{
|
|
var dataHandler = Composer.Instance.GetExportedValueByTypeName<IDataQueueHandler>(dataHandlerName);
|
|
dataHandler.SetJob(job);
|
|
_dataHandlers.Add(dataHandler);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns whether the data provider is connected
|
|
/// </summary>
|
|
/// <returns>true if the data provider is connected</returns>
|
|
public bool IsConnected => true;
|
|
|
|
/// <summary>
|
|
/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
|
|
/// </summary>
|
|
public void Dispose()
|
|
{
|
|
foreach (var dataHandler in _dataHandlers)
|
|
{
|
|
dataHandler.Dispose();
|
|
}
|
|
}
|
|
}
|
|
}
|