using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using MarketData.MarketDataModel; using MarketData.DataAccess; using MarketData.Utils; namespace MarketData.Helper { public class EarningsAnnouncementsMarketDataHelper : MarketDataHelperBase { private static int MaxThreads = 7; // (int)ThreadHelperEnum.MaxThreads; public EarningsAnnouncementsMarketDataHelper() { } public bool UpdateEarningsAnnouncements(String symbol=null) { Profiler profiler=new Profiler(); try { if(symbol==null)Queue=PricingDA.GetSymbols(); else (Queue=new List()).Add(symbol); Index=-1; ManualResetEvent[] resetEvents = new ManualResetEvent[MaxThreads]; for (int eventIndex = 0; eventIndex < resetEvents.Length; eventIndex++)resetEvents[eventIndex] = new ManualResetEvent(true); MDTrace.WriteLine(String.Format("Queuing earnings announcement fetches ...")); DateTime modified=DateTime.Now; while (true) { ManualResetEvent[] availableEvents=GetAvailableEvents(resetEvents); ManualResetEvent[] busyEvents=GetBusyEvents(resetEvents); if (null == PeekQueueItem() && 0==busyEvents.Length) { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Earnings announcement queue contains {0} items, busy events {1}, all done.",0,busyEvents.Length)); break; } for (int index = 0; index < availableEvents.Length; index++) { symbol=GetQueueItem(); if (null != symbol) { availableEvents[index].Reset(); ThreadHelper earningsAnnouncementsThreadHelper = new ThreadHelper(symbol,modified,availableEvents[index]); ThreadPool.QueueUserWorkItem(ThreadPoolCallbackEarningsAnnouncements, earningsAnnouncementsThreadHelper); } else { busyEvents=GetBusyEvents(resetEvents); if(busyEvents.Length!=availableEvents.Length) { ManualResetEvent[] resizedEvents=new ManualResetEvent[busyEvents.Length]; Array.Copy(busyEvents, resizedEvents,busyEvents.Length); resetEvents = resizedEvents; } break; } } // for MDTrace.WriteLine(LogLevel.DEBUG,"Earnings announcements waiting for free slots..."); if(resetEvents.Length>0)WaitHandle.WaitAny(resetEvents); if(null==PeekQueueItem())resetEvents=ResizeEvents(resetEvents); } // while MDTrace.WriteLine(LogLevel.DEBUG,"Earnings announcements completed."); return true; } finally { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateEarningsAnnouncements]End, total took {0}(ms)",profiler.End())); } } public void ThreadPoolCallbackEarningsAnnouncements(Object earningsAnnouncementsThreadHelperContext) { ThreadHelper threadHelper = (ThreadHelper)earningsAnnouncementsThreadHelperContext; try { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load earnings announcements, Thread {0} started for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); LoadEarningsAnnouncementsEx(threadHelper.Symbol,threadHelper.Modified); MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load earnings announcements, Thread {0} ended for {1}", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); } finally { threadHelper.ResetEvent.Set(); } } // Single threaded method, use for debugging a single symbol fetch without going through the queue public static void LoadEarningsAnnouncementsEx(String symbol,DateTime modified) { EarningsAnnouncements earningsAnnouncements=MarketDataHelper.GetEarningsAnnouncements(symbol); if(null==earningsAnnouncements||0==earningsAnnouncements.Count) { MDTrace.WriteLine(LogLevel.DEBUG,"Error loading earnings announcements for '"+symbol+"'"); return; } MDTrace.WriteLine(String.Format("[UpdateEarningsAnnouncements] Got {0} for {1}",earningsAnnouncements.Count,symbol)); if(!EarningsAnnouncementsDA.InsertEarningsAnnouncements(earningsAnnouncements))MDTrace.WriteLine(String.Format("[UpdateEarningsAnnouncements] Insert failed for {0}",symbol)); else MDTrace.WriteLine(String.Format("[UpdateEarningsAnnouncements] Insert succeeded for {0}",symbol)); } } }