From 11ceb8e013f64c96cf55da127434b60424f2d028 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 22 Nov 2024 22:38:09 -0600 Subject: [PATCH 1/6] Add EventPriority --- .../Infrastructures/Enums/EventPriority.cs | 8 +++ .../Infrastructures/Events/IEventPublisher.cs | 2 +- .../Infrastructures/Events/RedisPublisher.cs | 10 ++- .../Infrastructures/Events/RedisSubscriber.cs | 72 ++++++++++++------- 4 files changed, 63 insertions(+), 29 deletions(-) create mode 100644 src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs new file mode 100644 index 00000000..75c4402d --- /dev/null +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Enums/EventPriority.cs @@ -0,0 +1,8 @@ +namespace BotSharp.Abstraction.Infrastructures.Enums; + +public enum EventPriority +{ + Low = 1, + Medium = 2, + High = 3 +} diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs index 68ceebd2..a58adf59 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -10,7 +10,7 @@ public interface IEventPublisher /// Task BroadcastAsync(string channel, string message); - Task PublishAsync(string channel, string message); + Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium); Task ReDispatchAsync(string channel, int count = 10, string order = "asc"); diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index 2ff585d6..d991a45c 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Enums; using StackExchange.Redis; namespace BotSharp.Core.Infrastructures.Events; @@ -20,10 +21,13 @@ public class RedisPublisher : IEventPublisher await _subscriber.PublishAsync(channel, message); } - public async Task PublishAsync(string channel, string message) + public async Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium) { var db = _redis.GetDatabase(); + // convert to apporiate channel by priority + channel = $"{channel}-{priority}"; + if (CheckMessageExists(db, channel, "message", message)) { _logger.LogError($"The message already exists {channel} {message}"); @@ -41,7 +45,7 @@ public class RedisPublisher : IEventPublisher _logger.LogInformation($"Published message {channel} {message} ({messageId})"); } - private bool CheckMessageExists(IDatabase db, string streamName, string fieldName, string desiredValue) + private bool CheckMessageExists(IDatabase db, string channel, string fieldName, string desiredValue) { // Define the range to fetch all messages RedisValue start = "-"; // Start from the smallest ID @@ -49,7 +53,7 @@ public class RedisPublisher : IEventPublisher int count = 10; // Number of messages to retrieve // Fetch the latest 10 messages - var streamEntries = db.StreamRange(streamName, start, end, count, Order.Descending); + var streamEntries = db.StreamRange(channel, start, end, count, Order.Descending); if (streamEntries.Length == 0) { diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index bf6652b1..f35113fe 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -1,3 +1,4 @@ +using BotSharp.Abstraction.Infrastructures.Enums; using StackExchange.Redis; namespace BotSharp.Core.Infrastructures.Events; @@ -28,6 +29,52 @@ public class RedisSubscriber : IEventSubscriber { var db = _redis.GetDatabase(); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group); + + while (true) + { + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + { + continue; + } + + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + { + continue; + } + + await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); + } + } + + private async Task HandleGroupMessage(IDatabase db, string channel, string group, Func received) + { + var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); + foreach (var entry in entries) + { + _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); + await db.StreamAcknowledgeAsync(channel, group, entry.Id); + + try + { + await received(channel, entry.Values[0].Value); + + // Optionally delete the message to save space + await db.StreamDeleteAsync(channel, [entry.Id]); + } + catch (Exception ex) + { + _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}"); + } + } + + return entries.Length; + } + + private async Task CreateConsumerGroup(IDatabase db, string channel, string group) + { // Create the consumer group if it doesn't exist try { @@ -43,30 +90,5 @@ public class RedisSubscriber : IEventSubscriber _logger.LogError($"Error creating consumer group: '{group}' {ex.Message}"); throw; } - - while (true) - { - var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1); - foreach (var entry in entries) - { - _logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}"); - await db.StreamAcknowledgeAsync(channel, group, entry.Id); - - try - { - await received(channel, entry.Values[0].Value); - - // Optionally delete the message to save space - await db.StreamDeleteAsync(channel, [entry.Id]); - } - catch (Exception ex) - { - _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}"); - } - } - - await Task.Delay(Random.Shared.Next(1, 11) * 100); - } - } } From 597e26a1e070533e1405f1adfeaf17d46bf6b231 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sat, 23 Nov 2024 14:48:22 +0000 Subject: [PATCH 2/6] Allow to disable EventPriority --- .../Infrastructures/Events/IEventPublisher.cs | 2 +- .../Events/IEventSubscriber.cs | 2 +- .../Infrastructures/Events/RedisPublisher.cs | 7 ++-- .../Infrastructures/Events/RedisSubscriber.cs | 36 +++++++++++++------ 4 files changed, 32 insertions(+), 15 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs index a58adf59..f2276c13 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs @@ -10,7 +10,7 @@ public interface IEventPublisher /// Task BroadcastAsync(string channel, string message); - Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium); + Task PublishAsync(string channel, string message, EventPriority? priority = null); Task ReDispatchAsync(string channel, int count = 10, string order = "asc"); diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs index f295e54e..fa96bb35 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs @@ -4,5 +4,5 @@ public interface IEventSubscriber { Task SubscribeAsync(string channel, Func received); - Task SubscribeAsync(string channel, string group, Func received); + Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func received); } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index d991a45c..27cbf566 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -21,12 +21,15 @@ public class RedisPublisher : IEventPublisher await _subscriber.PublishAsync(channel, message); } - public async Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium) + public async Task PublishAsync(string channel, string message, EventPriority? priority = null) { var db = _redis.GetDatabase(); // convert to apporiate channel by priority - channel = $"{channel}-{priority}"; + if (priority != null) + { + channel = $"{channel}-{priority}"; + } if (CheckMessageExists(db, channel, "message", message)) { diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index f35113fe..b93c22a2 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -25,27 +25,41 @@ public class RedisSubscriber : IEventSubscriber }); } - public async Task SubscribeAsync(string channel, string group, Func received) + public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func received) { var db = _redis.GetDatabase(); - await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group); - await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group); - await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group); + if (priorityEnabled) + { + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group); + await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group); + } + else + { + await CreateConsumerGroup(db, channel, group); + } while (true) { - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + if (priorityEnabled) { - continue; - } + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + { + continue; + } - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + { + continue; + } + + await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); + } + else { - continue; + await HandleGroupMessage(db, channel, group, received); } - - await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); } } From e92b72ceb520f99fe3f81badc6112cdc9a683003 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sat, 23 Nov 2024 19:19:24 +0000 Subject: [PATCH 3/6] Handle exception SubscribeAsync --- .../Infrastructures/Events/RedisSubscriber.cs | 27 ++++++++++++------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index b93c22a2..67dba26d 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -42,23 +42,30 @@ public class RedisSubscriber : IEventSubscriber while (true) { - if (priorityEnabled) + try { - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + if (priorityEnabled) { - continue; - } + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) + { + continue; + } - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) + { + continue; + } + + await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); + } + else { - continue; + await HandleGroupMessage(db, channel, group, received); } - - await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); } - else + catch (Exception ex) { - await HandleGroupMessage(db, channel, group, received); + _logger.LogError($"Error processing message: {ex.Message}\r\n{ex}"); } } } From f722781be491899256b3427a19dee851db01d2e1 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sat, 23 Nov 2024 20:02:43 -0600 Subject: [PATCH 4/6] DragAndDrop --- .../Browsing/Enums/BroswerActionEnum.cs | 1 + .../Browsing/Models/ElementPosition.cs | 5 + .../Infrastructures/Events/RedisSubscriber.cs | 27 ++---- .../PlaywrightWebDriver.DoAction.cs | 92 +++++++++++++++++++ 4 files changed, 108 insertions(+), 17 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Browsing/Enums/BroswerActionEnum.cs b/src/Infrastructure/BotSharp.Abstraction/Browsing/Enums/BroswerActionEnum.cs index 8f1cee10..abdfd67b 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Browsing/Enums/BroswerActionEnum.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Browsing/Enums/BroswerActionEnum.cs @@ -7,4 +7,5 @@ public enum BroswerActionEnum Typing = 3, Hover = 4, Scroll = 5, + DragAndDrop = 6 } diff --git a/src/Infrastructure/BotSharp.Abstraction/Browsing/Models/ElementPosition.cs b/src/Infrastructure/BotSharp.Abstraction/Browsing/Models/ElementPosition.cs index fd95bd66..9a1ef0d3 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Browsing/Models/ElementPosition.cs +++ b/src/Infrastructure/BotSharp.Abstraction/Browsing/Models/ElementPosition.cs @@ -5,4 +5,9 @@ public class ElementPosition public float X { get; set; } = default!; public float Y { get; set; } = default!; + + public override string ToString() + { + return $"[{X}, {Y}]"; + } } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index 67dba26d..b93c22a2 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -42,30 +42,23 @@ public class RedisSubscriber : IEventSubscriber while (true) { - try + if (priorityEnabled) { - if (priorityEnabled) + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) { - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) - { - continue; - } - - if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) - { - continue; - } - - await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); + continue; } - else + + if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0) { - await HandleGroupMessage(db, channel, group, received); + continue; } + + await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received); } - catch (Exception ex) + else { - _logger.LogError($"Error processing message: {ex.Message}\r\n{ex}"); + await HandleGroupMessage(db, channel, group, received); } } } diff --git a/src/Plugins/BotSharp.Plugin.WebDriver/Drivers/PlaywrightDriver/PlaywrightWebDriver.DoAction.cs b/src/Plugins/BotSharp.Plugin.WebDriver/Drivers/PlaywrightDriver/PlaywrightWebDriver.DoAction.cs index f7884beb..40fec84e 100644 --- a/src/Plugins/BotSharp.Plugin.WebDriver/Drivers/PlaywrightDriver/PlaywrightWebDriver.DoAction.cs +++ b/src/Plugins/BotSharp.Plugin.WebDriver/Drivers/PlaywrightDriver/PlaywrightWebDriver.DoAction.cs @@ -79,10 +79,102 @@ public partial class PlaywrightWebDriver { await locator.HoverAsync(); } + else if (action.Action == BroswerActionEnum.DragAndDrop) + { + // Locate the element to drag + var box = await locator.BoundingBoxAsync(); + + if (box != null) + { + // Calculate start position + float startX = box.X + box.Width / 2; // Start at the center of the element + float startY = box.Y + box.Height / 2; + + // Drag offsets + float offsetX = action.Position.X; + // Move horizontally + if (action.Position.Y == 0) + { + // Perform drag-and-move + // Move mouse to the start position + var mouse = page.Mouse; + await mouse.MoveAsync(startX, startY); + await mouse.DownAsync(); + + // Move mouse smoothly in increments + var tracks = GetVelocityTrack(offsetX); + foreach (var track in tracks) + { + startX += track; + await page.Mouse.MoveAsync(startX, 0, new MouseMoveOptions + { + Steps = 3 + }); + } + + // Release mouse button + await mouse.UpAsync(); + } + else + { + throw new NotImplementedException(); + } + } + } if (action.WaitTime > 0) { await Task.Delay(1000 * action.WaitTime); } } + + public static List GetVelocityTrack(float distance) + { + // Initialize the track list to store the movement distances + List track = new List(); + + // Initialize variables + float current = 0; // Current position + float mid = distance * 4 / 5; // Deceleration threshold + float t = 0.2f; // Time interval + float v = 1; // Initial velocity + + // Generate the track + while (current < distance) + { + float a; // Acceleration + + // Determine acceleration based on position + if (current < mid) + { + a = 4; // Accelerate + } + else + { + a = -3; // Decelerate + } + + // Calculate new velocity + float v0 = v; + v = v0 + a * t; + + // Calculate the movement during this interval + float move = v0 * t + 0.5f * a * t * t; + + // Update current position + if (current + move > distance) + { + move = distance - current; + track.Add((int)Math.Round(move)); + break; + } + + current += move; + + // Add rounded movement to the track + track.Add((int)Math.Round(move)); + } + + return track; + } } From 0b98fe0dfe3478c83b1c29dbfdec21c78f95db80 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sun, 24 Nov 2024 09:10:03 -0600 Subject: [PATCH 5/6] delay 100ms --- .../BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index b93c22a2..89b38a8d 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs @@ -42,6 +42,8 @@ public class RedisSubscriber : IEventSubscriber while (true) { + await Task.Delay(100); + if (priorityEnabled) { if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0) From 28908d9592544e70e2ecf3c1b88a1ead2ab2c9d5 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sun, 24 Nov 2024 11:04:22 -0600 Subject: [PATCH 6/6] disable lock --- .../BotSharp.Core/Infrastructures/Events/RedisPublisher.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index 27cbf566..435451d4 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs @@ -91,7 +91,10 @@ public class RedisPublisher : IEventPublisher try { - var messageId = await db.StreamAddAsync(channel, "message", entry.Values[0].Value); + var messageId = await db.StreamAddAsync(channel, [ + new NameValueEntry("message", entry.Values[0].Value), + new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")) + ]); _logger.LogWarning($"ReDispatched message: {channel} {entry.Values[0].Value} ({messageId})");