Merge pull request #1118 from karashiiro/feat/rxnet-improvements

Dalamud/Game/Network/Internal/MarketBoardUploaders/MarketBoardItemRequest.cs

@@ -20,6 +20,16 @@ internal class MarketBoardItemRequest
     /// </summary>
     public uint CatalogId { get; private set; }
 
+    /// <summary>
+    /// Gets the request status. Nonzero statuses are errors.
+    /// </summary>
+    public uint Status { get; private set; }
+
+    /// <summary>
+    /// Gets a value indicating whether or not this request was successful.
+    /// </summary>
+    public bool Ok => this.Status == 0;
+
     /// <summary>
     /// Gets the amount to arrive.
     /// </summary>
@@ -58,7 +68,8 @@ internal class MarketBoardItemRequest
         var output = new MarketBoardItemRequest();
 
         output.CatalogId = reader.ReadUInt32();
-        stream.Position += 0x7;
+        output.Status = reader.ReadUInt32();
+        stream.Position += 0x3;
         output.AmountToArrive = reader.ReadByte();
 
         return output;

Dalamud/Game/Network/Internal/MarketBoardUploaders/Universalis/Types/UniversalisItemUploadRequest.cs

@@ -7,7 +7,7 @@ namespace Dalamud.Game.Network.Internal.MarketBoardUploaders.Universalis.Types;
 /// <summary>
 /// A Universalis API structure.
 /// </summary>
-internal class UniversalisHistoryUploadRequest
+internal class UniversalisItemUploadRequest
 {
     /// <summary>
     /// Gets or sets the world ID.
@@ -21,11 +21,17 @@ internal class UniversalisHistoryUploadRequest
     [JsonProperty("itemID")]
     public uint ItemId { get; set; }
 
+    /// <summary>
+    /// Gets or sets the list of available items.
+    /// </summary>
+    [JsonProperty("listings")]
+    public List<UniversalisItemListingsEntry> Listings { get; set; }
+
     /// <summary>
     /// Gets or sets the list of available entries.
     /// </summary>
     [JsonProperty("entries")]
-    public List<UniversalisHistoryEntry> Entries { get; set; }
+    public List<UniversalisHistoryEntry> Sales { get; set; }
 
     /// <summary>
     /// Gets or sets the uploader ID.

Dalamud/Game/Network/Internal/MarketBoardUploaders/Universalis/UniversalisMarketBoardUploader.cs

@@ -41,12 +41,13 @@ internal class UniversalisMarketBoardUploader : IMarketBoardUploader
 
         // ====================================================================================
 
-        var listingsUploadObject = new UniversalisItemListingsUploadRequest
+        var uploadObject = new UniversalisItemUploadRequest
         {
             WorldId = clientState.LocalPlayer?.CurrentWorld.Id ?? 0,
             UploaderId = uploader.ToString(),
             ItemId = request.CatalogId,
             Listings = new List<UniversalisItemListingsEntry>(),
+            Sales = new List<UniversalisHistoryEntry>(),
         };
 
         foreach (var marketBoardItemListing in request.Listings)
@@ -77,27 +78,12 @@ internal class UniversalisMarketBoardUploader : IMarketBoardUploader
                 });
             }
 
-            listingsUploadObject.Listings.Add(universalisListing);
+            uploadObject.Listings.Add(universalisListing);
         }
 
-        var listingPath = "/upload";
-        var listingUpload = JsonConvert.SerializeObject(listingsUploadObject);
-        Log.Verbose("{ListingPath}: {ListingUpload}", listingPath, listingUpload);
-        await Util.HttpClient.PostAsync($"{ApiBase}{listingPath}/{ApiKey}", new StringContent(listingUpload, Encoding.UTF8, "application/json"));
-
-        // ====================================================================================
-
-        var historyUploadObject = new UniversalisHistoryUploadRequest
-        {
-            WorldId = clientState.LocalPlayer?.CurrentWorld.Id ?? 0,
-            UploaderId = uploader.ToString(),
-            ItemId = request.CatalogId,
-            Entries = new List<UniversalisHistoryEntry>(),
-        };
-
         foreach (var marketBoardHistoryListing in request.History)
         {
-            historyUploadObject.Entries.Add(new UniversalisHistoryEntry
+            uploadObject.Sales.Add(new UniversalisHistoryEntry
             {
                 BuyerName = marketBoardHistoryListing.BuyerName,
                 Hq = marketBoardHistoryListing.IsHq,
@@ -108,10 +94,10 @@ internal class UniversalisMarketBoardUploader : IMarketBoardUploader
             });
         }
 
-        var historyPath = "/upload";
+        var uploadPath = "/upload";
-        var historyUpload = JsonConvert.SerializeObject(historyUploadObject);
+        var uploadData = JsonConvert.SerializeObject(uploadObject);
-        Log.Verbose("{HistoryPath}: {HistoryUpload}", historyPath, historyUpload);
+        Log.Verbose("{ListingPath}: {ListingUpload}", uploadPath, uploadData);
-        await Util.HttpClient.PostAsync($"{ApiBase}{historyPath}/{ApiKey}", new StringContent(historyUpload, Encoding.UTF8, "application/json"));
+        await Util.HttpClient.PostAsync($"{ApiBase}{uploadPath}/{ApiKey}", new StringContent(uploadData, Encoding.UTF8, "application/json"));
 
         // ====================================================================================
 

Dalamud/Game/Network/Internal/NetworkHandlers.cs

@@ -2,8 +2,6 @@ using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
-using System.Linq;
-using System.Reactive;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using System.Runtime.InteropServices;
@@ -29,39 +27,29 @@ internal class NetworkHandlers : IDisposable, IServiceType
 {
     private readonly IMarketBoardUploader uploader;
 
-    private readonly List<MarketBoardItemRequest> marketBoardRequests;
     private readonly ISubject<NetworkMessage> messages;
 
     private readonly IDisposable handleMarketBoardItemRequest;
-    private readonly IDisposable handleMarketBoardOfferings;
-    private readonly IDisposable handleMarketBoardHistory;
     private readonly IDisposable handleMarketTaxRates;
     private readonly IDisposable handleMarketBoardPurchaseHandler;
-    private readonly IDisposable handleMarketBoardPurchase;
     private readonly IDisposable handleCfPop;
 
     [ServiceManager.ServiceDependency]
     private readonly DalamudConfiguration configuration = Service<DalamudConfiguration>.Get();
 
-    private MarketBoardPurchaseHandler? marketBoardPurchaseHandler;
-
     private bool disposing;
 
     [ServiceManager.ServiceConstructor]
     private NetworkHandlers(GameNetwork gameNetwork)
     {
         this.uploader = new UniversalisMarketBoardUploader();
-        this.marketBoardRequests = new List<MarketBoardItemRequest>();
         this.CfPop = (_, _) => { };
 
         this.messages = new Subject<NetworkMessage>();
 
         this.handleMarketBoardItemRequest = this.HandleMarketBoardItemRequest();
-        this.handleMarketBoardOfferings = this.HandleMarketBoardOfferings();
-        this.handleMarketBoardHistory = this.HandleMarketBoardHistory();
         this.handleMarketTaxRates = this.HandleMarketTaxRates();
         this.handleMarketBoardPurchaseHandler = this.HandleMarketBoardPurchaseHandler();
-        this.handleMarketBoardPurchase = this.HandleMarketBoardPurchase();
         this.handleCfPop = this.HandleCfPop();
 
         gameNetwork.NetworkMessage += this.ObserveNetworkMessage;
@@ -91,11 +79,8 @@ internal class NetworkHandlers : IDisposable, IServiceType
             return;
 
         this.handleMarketBoardItemRequest.Dispose();
-        this.handleMarketBoardOfferings.Dispose();
-        this.handleMarketBoardHistory.Dispose();
         this.handleMarketTaxRates.Dispose();
         this.handleMarketBoardPurchaseHandler.Dispose();
-        this.handleMarketBoardPurchase.Dispose();
         this.handleCfPop.Dispose();
     }
 
@@ -119,7 +104,7 @@ internal class NetworkHandlers : IDisposable, IServiceType
         return this.messages.Where(message => message.DataManager?.IsDataReady == true);
     }
 
-    private IObservable<MarketBoardItemRequest> OnMarketBoardItemRequest()
+    private IObservable<MarketBoardItemRequest> OnMarketBoardItemRequestStart()
     {
         return this.OnNetworkMessage()
                    .Where(message => message.Direction == NetworkMessageDirection.ZoneDown)
@@ -182,244 +167,223 @@ internal class NetworkHandlers : IDisposable, IServiceType
                    .Where(message => message.Opcode == message.DataManager?.ServerOpCodes["CfNotifyPop"]);
     }
 
+    private IObservable<List<MarketBoardCurrentOfferings.MarketBoardItemListing>> OnMarketBoardListingsBatch(
+        IObservable<MarketBoardItemRequest> start)
+    {
+        var startShared = start.Publish().RefCount();
+        var offeringsObservable = this.OnMarketBoardOfferings().Publish().RefCount();
+
+        void LogStartObserved(MarketBoardItemRequest request)
+        {
+            Log.Verbose(
+                "Observed start of request for item#{CatalogId} with {NumListings} expected listings",
+                request.CatalogId,
+                request.AmountToArrive);
+        }
+
+        void LogEndObserved(MarketBoardCurrentOfferings offerings)
+        {
+            Log.Verbose(
+                "Observed end of request {RequestId}",
+                offerings.RequestId);
+        }
+
+        void LogOfferingsObserved(MarketBoardCurrentOfferings offerings)
+        {
+            Log.Verbose(
+                "Observed element of request {RequestId} with {NumListings} listings",
+                offerings.RequestId,
+                offerings.ItemListings.Count);
+        }
+
+        IObservable<MarketBoardCurrentOfferings> UntilBatchEnd(MarketBoardItemRequest request)
+        {
+            var totalPackets = Convert.ToInt32(Math.Ceiling((double)request.AmountToArrive / 10));
+            return offeringsObservable
+                   .Skip(totalPackets - 1)
+                   .Do(LogEndObserved);
+        }
+
+        // When a start packet is observed, begin observing a window of listings packets
+        // according to the count described by the start packet. Aggregate the listings
+        // packets, and then flatten them to the listings themselves.
+        return offeringsObservable
+               .Do(LogOfferingsObserved)
+               .Window(startShared.Where(request => request.Ok).Do(LogStartObserved), UntilBatchEnd)
+               .SelectMany(
+                   o => o.Aggregate(
+                       new List<MarketBoardCurrentOfferings.MarketBoardItemListing>(),
+                       (agg, next) =>
+                       {
+                           agg.AddRange(next.ItemListings);
+                           return agg;
+                       }));
+    }
+
+    private IObservable<List<MarketBoardHistory.MarketBoardHistoryListing>> OnMarketBoardSalesBatch()
+    {
+        return this.OnMarketBoardHistory().Select(history => history.HistoryListings);
+    }
+
     private IDisposable HandleMarketBoardItemRequest()
     {
-        return this.OnMarketBoardItemRequest()
+        var startObservable = this.OnMarketBoardItemRequestStart();
-                   .Where(_ => this.configuration.IsMbCollect)
+        return Observable.When(
-                   .Subscribe(request =>
+                             startObservable
-                   {
+                                 .And(this.OnMarketBoardSalesBatch())
-                       this.marketBoardRequests.Add(request);
+                                 .And(this.OnMarketBoardListingsBatch(startObservable))
-                       Log.Verbose($"NEW MB REQUEST START: item#{request.CatalogId} amount#{request.AmountToArrive}");
+                                 .Then((request, sales, listings) => (request, sales, listings)))
-                   });
+                         .Where(this.ShouldUpload)
+                         .Subscribe(
+                             data =>
+                             {
+                                 var (request, sales, listings) = data;
+                                 this.UploadMarketBoardData(request, sales, listings);
+                             },
+                             ex => Log.Error(ex, "Failed to handle Market Board item request event"));
     }
 
-    private IDisposable HandleMarketBoardOfferings()
+    private void UploadMarketBoardData(
+        MarketBoardItemRequest request,
+        ICollection<MarketBoardHistory.MarketBoardHistoryListing> sales,
+        ICollection<MarketBoardCurrentOfferings.MarketBoardItemListing> listings)
     {
-        return this.OnMarketBoardOfferings()
+        Log.Verbose(
-                   .Where(_ => this.configuration.IsMbCollect)
+            "Market Board request resolved, starting upload: item#{CatalogId} listings#{ListingsObserved} sales#{SalesObserved}",
-                   .Subscribe(listing =>
+            request.CatalogId,
-                   {
+            listings.Count,
-                       var request =
+            sales.Count);
-                           this.marketBoardRequests.LastOrDefault(
-                               r => r.CatalogId == listing.ItemListings[0].CatalogId && !r.IsDone);
 
-                       if (request == default)
+        request.Listings.AddRange(listings);
-                       {
+        request.History.AddRange(sales);
-                           Log.Error(
-                               $"Market Board data arrived without a corresponding request: item#{listing.ItemListings[0].CatalogId}");
-                           return;
-                       }
 
-                       if (request.Listings.Count + listing.ItemListings.Count > request.AmountToArrive)
+        Task.Run(() => this.uploader.Upload(request))
-                       {
+            .ContinueWith(
-                           Log.Error(
+                task => Log.Error(task.Exception, "Market Board offerings data upload failed"),
-                               $"Too many Market Board listings received for request: {request.Listings.Count + listing.ItemListings.Count} > {request.AmountToArrive} item#{listing.ItemListings[0].CatalogId}");
+                TaskContinuationOptions.OnlyOnFaulted);
-                           return;
-                       }
-
-                       if (request.ListingsRequestId != -1 && request.ListingsRequestId != listing.RequestId)
-                       {
-                           Log.Error(
-                               $"Non-matching RequestIds for Market Board data request: {request.ListingsRequestId}, {listing.RequestId}");
-                           return;
-                       }
-
-                       if (request.ListingsRequestId == -1 && request.Listings.Count > 0)
-                       {
-                           Log.Error(
-                               $"Market Board data request sequence break: {request.ListingsRequestId}, {request.Listings.Count}");
-                           return;
-                       }
-
-                       if (request.ListingsRequestId == -1)
-                       {
-                           request.ListingsRequestId = listing.RequestId;
-                           Log.Verbose($"First Market Board packet in sequence: {listing.RequestId}");
-                       }
-
-                       request.Listings.AddRange(listing.ItemListings);
-
-                       Log.Verbose(
-                           "Added {0} ItemListings to request#{1}, now {2}/{3}, item#{4}",
-                           listing.ItemListings.Count,
-                           request.ListingsRequestId,
-                           request.Listings.Count,
-                           request.AmountToArrive,
-                           request.CatalogId);
-
-                       if (request.IsDone)
-                       {
-                           Log.Verbose(
-                               "Market Board request finished, starting upload: request#{0} item#{1} amount#{2}",
-                               request.ListingsRequestId,
-                               request.CatalogId,
-                               request.AmountToArrive);
-
-                           Task.Run(() => this.uploader.Upload(request))
-                               .ContinueWith(
-                                   task => Log.Error(task.Exception, "Market Board offerings data upload failed."),
-                                   TaskContinuationOptions.OnlyOnFaulted);
-                       }
-                   });
-    }
-
-    private IDisposable HandleMarketBoardHistory()
-    {
-        return this.OnMarketBoardHistory()
-                   .Where(_ => this.configuration.IsMbCollect)
-                   .Subscribe(listing =>
-                   {
-                       var request = this.marketBoardRequests.LastOrDefault(r => r.CatalogId == listing.CatalogId);
-
-                       if (request == default)
-                       {
-                           Log.Error(
-                               $"Market Board data arrived without a corresponding request: item#{listing.CatalogId}");
-                           return;
-                       }
-
-                       if (request.ListingsRequestId != -1)
-                       {
-                           Log.Error(
-                               $"Market Board data history sequence break: {request.ListingsRequestId}, {request.Listings.Count}");
-                           return;
-                       }
-
-                       request.History.AddRange(listing.HistoryListings);
-
-                       Log.Verbose("Added history for item#{0}", listing.CatalogId);
-
-                       if (request.AmountToArrive == 0)
-                       {
-                           Log.Verbose("Request had 0 amount, uploading now");
-
-                           Task.Run(() => this.uploader.Upload(request))
-                               .ContinueWith(
-                                   (task) => Log.Error(task.Exception, "Market Board history data upload failed."),
-                                   TaskContinuationOptions.OnlyOnFaulted);
-                       }
-                   });
     }
 
     private IDisposable HandleMarketTaxRates()
     {
         return this.OnMarketTaxRates()
-                   .Where(_ => this.configuration.IsMbCollect)
+                   .Where(this.ShouldUpload)
-                   .Subscribe(taxes =>
+                   .Subscribe(
-                   {
+                       taxes =>
-                       Log.Verbose(
+                       {
-                           "MarketTaxRates: limsa#{0} grid#{1} uldah#{2} ish#{3} kugane#{4} cr#{5} sh#{6}",
+                           Log.Verbose(
-                           taxes.LimsaLominsaTax,
+                               "MarketTaxRates: limsa#{0} grid#{1} uldah#{2} ish#{3} kugane#{4} cr#{5} sh#{6}",
-                           taxes.GridaniaTax,
+                               taxes.LimsaLominsaTax,
-                           taxes.UldahTax,
+                               taxes.GridaniaTax,
-                           taxes.IshgardTax,
+                               taxes.UldahTax,
-                           taxes.KuganeTax,
+                               taxes.IshgardTax,
-                           taxes.CrystariumTax,
+                               taxes.KuganeTax,
-                           taxes.SharlayanTax);
+                               taxes.CrystariumTax,
+                               taxes.SharlayanTax);
 
-                       Task.Run(() => this.uploader.UploadTax(taxes))
+                           Task.Run(() => this.uploader.UploadTax(taxes))
-                           .ContinueWith(
+                               .ContinueWith(
-                               task => Log.Error(task.Exception, "Market Board tax data upload failed."),
+                                   task => Log.Error(task.Exception, "Market Board tax data upload failed"),
-                               TaskContinuationOptions.OnlyOnFaulted);
+                                   TaskContinuationOptions.OnlyOnFaulted);
-                   });
+                       },
+                       ex => Log.Error(ex, "Failed to handle Market Board tax data event"));
     }
 
     private IDisposable HandleMarketBoardPurchaseHandler()
     {
         return this.OnMarketBoardPurchaseHandler()
-                   .Where(_ => this.configuration.IsMbCollect)
+                   .Zip(this.OnMarketBoardPurchase())
-                   .Subscribe(handler => { this.marketBoardPurchaseHandler = handler; });
+                   .Where(this.ShouldUpload)
-    }
+                   .Subscribe(
-
+                       data =>
-    private IDisposable HandleMarketBoardPurchase()
-    {
-        return this.OnMarketBoardPurchase()
-                   .Where(_ => this.configuration.IsMbCollect)
-                   .Subscribe(purchase =>
-                   {
-                       if (this.marketBoardPurchaseHandler == null)
-                           return;
-
-                       var sameQty = purchase.ItemQuantity == this.marketBoardPurchaseHandler.ItemQuantity;
-                       var itemMatch = purchase.CatalogId == this.marketBoardPurchaseHandler.CatalogId;
-                       var itemMatchHq = purchase.CatalogId == this.marketBoardPurchaseHandler.CatalogId + 1_000_000;
-
-                       // Transaction succeeded
-                       if (sameQty && (itemMatch || itemMatchHq))
                        {
-                           Log.Verbose(
+                           var (handler, purchase) = data;
-                               $"Bought {purchase.ItemQuantity}x {this.marketBoardPurchaseHandler.CatalogId} for {this.marketBoardPurchaseHandler.PricePerUnit * purchase.ItemQuantity} gils, listing id is {this.marketBoardPurchaseHandler.ListingId}");
 
-                           var handler =
+                           var sameQty = purchase.ItemQuantity == handler.ItemQuantity;
-                               this.marketBoardPurchaseHandler; // Capture the object so that we don't pass in a null one when the task starts.
+                           var itemMatch = purchase.CatalogId == handler.CatalogId;
+                           var itemMatchHq = purchase.CatalogId == handler.CatalogId + 1_000_000;
 
-                           Task.Run(() => this.uploader.UploadPurchase(handler))
+                           // Transaction succeeded
-                               .ContinueWith(
+                           if (sameQty && (itemMatch || itemMatchHq))
-                                   task => Log.Error(task.Exception, "Market Board purchase data upload failed."),
+                           {
-                                   TaskContinuationOptions.OnlyOnFaulted);
+                               Log.Verbose(
-                       }
+                                   "Bought {PurchaseItemQuantity}x {HandlerCatalogId} for {HandlerPricePerUnit} gils, listing id is {HandlerListingId}",
-
+                                   purchase.ItemQuantity,
-                       this.marketBoardPurchaseHandler = null;
+                                   handler.CatalogId,
-                   });
+                                   handler.PricePerUnit * purchase.ItemQuantity,
+                                   handler.ListingId);
+                               Task.Run(() => this.uploader.UploadPurchase(handler))
+                                   .ContinueWith(
+                                       task => Log.Error(task.Exception, "Market Board purchase data upload failed"),
+                                       TaskContinuationOptions.OnlyOnFaulted);
+                           }
+                       },
+                       ex => Log.Error(ex, "Failed to handle Market Board purchase event"));
     }
 
     private unsafe IDisposable HandleCfPop()
     {
         return this.OnCfNotifyPop()
-                   .Subscribe(message =>
+                   .Subscribe(
-                   {
+                       message =>
-                       using var stream = new UnmanagedMemoryStream((byte*)message.Data.ToPointer(), 64);
-                       using var reader = new BinaryReader(stream);
-
-                       var notifyType = reader.ReadByte();
-                       stream.Position += 0x1B;
-                       var conditionId = reader.ReadUInt16();
-
-                       if (notifyType != 3)
-                           return;
-
-                       var cfConditionSheet = message.DataManager!.GetExcelSheet<ContentFinderCondition>()!;
-                       var cfCondition = cfConditionSheet.GetRow(conditionId);
-
-                       if (cfCondition == null)
                        {
-                           Log.Error($"CFC key {conditionId} not in Lumina data.");
+                           using var stream = new UnmanagedMemoryStream((byte*)message.Data.ToPointer(), 64);
-                           return;
+                           using var reader = new BinaryReader(stream);
-                       }
 
-                       var cfcName = cfCondition.Name.ToString();
+                           var notifyType = reader.ReadByte();
-                       if (cfcName.IsNullOrEmpty())
+                           stream.Position += 0x1B;
-                       {
+                           var conditionId = reader.ReadUInt16();
-                           cfcName = "Duty Roulette";
-                           cfCondition.Image = 112324;
-                       }
 
-                       // Flash window
+                           if (notifyType != 3)
-                       if (this.configuration.DutyFinderTaskbarFlash && !NativeFunctions.ApplicationIsActivated())
+                               return;
-                       {
+
-                           var flashInfo = new NativeFunctions.FlashWindowInfo
+                           var cfConditionSheet = message.DataManager!.GetExcelSheet<ContentFinderCondition>()!;
+                           var cfCondition = cfConditionSheet.GetRow(conditionId);
+
+                           if (cfCondition == null)
                            {
-                               Size = (uint)Marshal.SizeOf<NativeFunctions.FlashWindowInfo>(),
+                               Log.Error("CFC key {ConditionId} not in Lumina data", conditionId);
-                               Count = uint.MaxValue,
+                               return;
-                               Timeout = 0,
-                               Flags = NativeFunctions.FlashWindow.All | NativeFunctions.FlashWindow.TimerNoFG,
-                               Hwnd = Process.GetCurrentProcess().MainWindowHandle,
-                           };
-                           NativeFunctions.FlashWindowEx(ref flashInfo);
-                       }
-
-                       Task.Run(() =>
-                       {
-                           if (this.configuration.DutyFinderChatMessage)
-                           {
-                               Service<ChatGui>.GetNullable()?.Print($"Duty pop: {cfcName}");
                            }
 
-                           this.CfPop.InvokeSafely(this, cfCondition);
+                           var cfcName = cfCondition.Name.ToString();
-                       }).ContinueWith(
+                           if (cfcName.IsNullOrEmpty())
-                           task => Log.Error(task.Exception, "CfPop.Invoke failed."),
+                           {
-                           TaskContinuationOptions.OnlyOnFaulted);
+                               cfcName = "Duty Roulette";
-                   });
+                               cfCondition.Image = 112324;
+                           }
+
+                           // Flash window
+                           if (this.configuration.DutyFinderTaskbarFlash && !NativeFunctions.ApplicationIsActivated())
+                           {
+                               var flashInfo = new NativeFunctions.FlashWindowInfo
+                               {
+                                   Size = (uint)Marshal.SizeOf<NativeFunctions.FlashWindowInfo>(),
+                                   Count = uint.MaxValue,
+                                   Timeout = 0,
+                                   Flags = NativeFunctions.FlashWindow.All | NativeFunctions.FlashWindow.TimerNoFG,
+                                   Hwnd = Process.GetCurrentProcess().MainWindowHandle,
+                               };
+                               NativeFunctions.FlashWindowEx(ref flashInfo);
+                           }
+
+                           Task.Run(() =>
+                           {
+                               if (this.configuration.DutyFinderChatMessage)
+                               {
+                                   Service<ChatGui>.GetNullable()?.Print($"Duty pop: {cfcName}");
+                               }
+
+                               this.CfPop.InvokeSafely(this, cfCondition);
+                           }).ContinueWith(
+                               task => Log.Error(task.Exception, "CfPop.Invoke failed"),
+                               TaskContinuationOptions.OnlyOnFaulted);
+                       },
+                       ex => Log.Error(ex, "Failed to handle Market Board purchase event"));
+    }
+
+    private bool ShouldUpload<T>(T any)
+    {
+        return this.configuration.IsMbCollect;
     }
 
     private class NetworkMessage