diff --git a/29WxOrderWpf/MainWindow.xaml.cs b/29WxOrderWpf/MainWindow.xaml.cs index cb61924..a74ec32 100644 --- a/29WxOrderWpf/MainWindow.xaml.cs +++ b/29WxOrderWpf/MainWindow.xaml.cs @@ -106,10 +106,10 @@ namespace _29WxOrderWpf pddBuyerOrderReadCtrl.Visibility = Visibility.Visible; tbBuyerOrderReadCtrl.Visibility = Visibility.Visible; pddBuyerSelectCtrl.Visibility = Visibility.Visible; - Task.Run(() => + Task.Run(async () => { - GetServerShop(); - StartTicketRealtimeMonitor();//启动工单监控 + GetServerShop(); + await StartTicketRealtimeMonitor();//启动工单监控 GetShopSupplier(GlobalConfig.ShopLoginList.ToList()); }); Task.Run(() => @@ -609,12 +609,12 @@ namespace _29WxOrderWpf _ticketMonitor.MarkAllRead(); } - private void StartTicketRealtimeMonitor() + private async Task StartTicketRealtimeMonitor() { try { var options = GetTicketOptions().Clone(); - _ticketMonitor.Start(options); + await _ticketMonitor.Start(options); UpdateOrderSupportBadge(); _ = LoadHistoricalTicketUnreadAsync(options); } @@ -637,7 +637,7 @@ namespace _29WxOrderWpf } catch (Exception ex) { - LogSave.Add("TicketUnreadLoadError:" + ex); + NLogger.Instance.Error("TicketUnreadLoadError:" + ex); } } @@ -652,9 +652,6 @@ namespace _29WxOrderWpf AppIds = GlobalConfig.ShopLoginList.Select(a => a.ShopId).ToList(), Orders = [] }; - - - //OrderSupportTicketView // .CreateDefaultOptions(BaseClass.UserName) // .Clone(); diff --git a/AllBase/BaseClass.cs b/AllBase/BaseClass.cs index 47a5aed..e888177 100644 --- a/AllBase/BaseClass.cs +++ b/AllBase/BaseClass.cs @@ -37,7 +37,7 @@ namespace AllBase /// 版本ID /// - public static int VersionCount = 337; + public static int VersionCount = 339; public static string ServerUrl { get; set; } = "http://1.13.195.131:9000"; public static string DebugUrl { get; set; } = "http://localhost:9000"; diff --git a/OrderSupportTicket/LaunchDarkly.EventSource/Configuration.cs b/OrderSupportTicket/LaunchDarkly.EventSource/Configuration.cs index 5bc1aaa..bc44ea7 100644 --- a/OrderSupportTicket/LaunchDarkly.EventSource/Configuration.cs +++ b/OrderSupportTicket/LaunchDarkly.EventSource/Configuration.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Net.Http; using System.Text; +using WxOrder.Framework.Helpers; namespace LaunchDarkly.EventSource { @@ -114,6 +115,7 @@ namespace LaunchDarkly.EventSource /// This is never null; if logging is not configured, it will be LaunchDarkly.Logging.Logs.None. /// /// + public ILogger Logger { get; } /// /// The maximum amount of time to wait before attempting to reconnect. @@ -174,13 +176,14 @@ namespace LaunchDarkly.EventSource { Uri = builder._uri; - + var logger = builder._logger; BackoffResetThreshold = builder._backoffResetThreshold; HttpClient = builder._httpClient; HttpMessageHandler = (builder._httpClient != null) ? null : builder._httpMessageHandler; InitialRetryDelay = builder._initialRetryDelay; LastEventId = builder._lastEventId; + Logger = logger; MaxRetryDelay = builder._maxRetryDelay; Method = builder._method; PreferDataAsUtf8Bytes = builder._preferDataAsUtf8Bytes; diff --git a/OrderSupportTicket/LaunchDarkly.EventSource/ConfigurationBuilder.cs b/OrderSupportTicket/LaunchDarkly.EventSource/ConfigurationBuilder.cs index 6915a7a..dd06c36 100644 --- a/OrderSupportTicket/LaunchDarkly.EventSource/ConfigurationBuilder.cs +++ b/OrderSupportTicket/LaunchDarkly.EventSource/ConfigurationBuilder.cs @@ -3,6 +3,7 @@ using System.Collections.Generic; using System.Net.Http; using System.Text; using System.Threading; +using WxOrder.Framework.Helpers; namespace LaunchDarkly.EventSource { @@ -30,6 +31,7 @@ namespace LaunchDarkly.EventSource internal TimeSpan _initialRetryDelay = Configuration.DefaultInitialRetryDelay; internal TimeSpan _backoffResetThreshold = Configuration.DefaultBackoffResetThreshold; internal string _lastEventId; + internal ILogger _logger; internal IDictionary _requestHeaders = new Dictionary(); internal HttpMessageHandler _httpMessageHandler; internal HttpClient _httpClient; @@ -230,45 +232,6 @@ namespace LaunchDarkly.EventSource return this; } - /// - /// Sets the logging implementation to be used for all EventSource log output. - /// - /// - /// - /// This uses the ILogAdapter abstraction from the LaunchDarkly.Logging library, - /// which provides several basic implementations such as Logs.ToConsole and an integration - /// with the .NET Core logging framework. For more about this and about adapters to other logging - /// frameworks, see LaunchDarkly.Logging. - /// - /// - /// LaunchDarkly.Logging defines logging levels of Debug, Info, Warn, and Error. If you do not - /// want detailed Debug-level logging, use the Level() modifier to set a minimum level of Info - /// or above, as shown in the code example (unless you are using an adapter to another logging - /// framework that has its own way of doing log filtering). - /// - /// - /// Log messages will use as the logger name. If you - /// want to specify a different logger name, use . - /// - /// - /// If you don't specify or , - /// EventSource will not do any logging. - /// - /// - /// - /// using LaunchDarkly.Logging; - /// - /// // Send log output to the console (standard error), suppressing Debug messages - /// var config = new ConfigurationBuilder(uri). - /// LogAdapter(Logs.ToConsole.Level(LogLevel.Info)). - /// Build(); - /// - /// a LaunchDarkly.Logging.ILogAdapter - /// the builder - public ConfigurationBuilder LogAdapter() - { - return this; - } /// /// Sets a custom logger to be used for all EventSource log output. @@ -297,8 +260,9 @@ namespace LaunchDarkly.EventSource /// /// a LaunchDarkly.Logging.Logger instance /// the builder - public ConfigurationBuilder Logger() + public ConfigurationBuilder Logger(ILogger logger) { + _logger = logger; return this; } diff --git a/OrderSupportTicket/LaunchDarkly.EventSource/EventSource.cs b/OrderSupportTicket/LaunchDarkly.EventSource/EventSource.cs index 5d1dd7d..51516bd 100644 --- a/OrderSupportTicket/LaunchDarkly.EventSource/EventSource.cs +++ b/OrderSupportTicket/LaunchDarkly.EventSource/EventSource.cs @@ -5,504 +5,514 @@ using System.IO; using System.Net.Http; using System.Threading; using System.Threading.Tasks; +using WxOrder.Framework.Helpers; namespace LaunchDarkly.EventSource { - /// - /// Provides an EventSource client for consuming Server-Sent Events. Additional details on the Server-Sent Events spec - /// can be found at https://html.spec.whatwg.org/multipage/server-sent-events.html - /// - public class EventSource : IEventSource, IDisposable - { - #region Private Fields - - private readonly Configuration _configuration; - private readonly HttpClient _httpClient; - - private List _eventDataStringBuffer; - private MemoryStream _eventDataUtf8ByteBuffer; - private string _eventName; - private string _lastEventId; - private readonly ExponentialBackoffWithDecorrelation _backOff; - private readonly TimeSpan _configuredInitialRetryDelay; - private readonly TimeSpan _configuredMaxRetryDelay; - private CancellationTokenSource _currentRequestToken; - private readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource(); - private static readonly TimeSpan MaxSleepTime = TimeSpan.FromMilliseconds(int.MaxValue); - private readonly Stopwatch _connectionTimer = new Stopwatch(); - private ReadyState _readyState; - - #endregion - - #region Public Events - - /// - public event EventHandler Opened; - - /// - public event EventHandler Closed; - - /// - public event EventHandler MessageReceived; - - /// - public event EventHandler CommentReceived; - - /// - public event EventHandler Error; - - #endregion Public Events - - #region Public Properties - - /// - public ReadyState ReadyState - { - get - { - lock (this) - { - return _readyState; - } - } - private set - { - lock (this) - { - _readyState = value; - } - } - } - - internal TimeSpan BackOffDelay - { - get; - private set; - } - - // Exposed for tests, so that retry-bounds behavior can be asserted directly - internal ExponentialBackoffWithDecorrelation BackOff => _backOff; - - #endregion - - #region Public Constructors - - /// - /// Initializes a new instance of the class. - /// - /// the configuration - public EventSource(Configuration configuration) - { - _readyState = ReadyState.Raw; - - _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); - - - _configuredInitialRetryDelay = _configuration.InitialRetryDelay; - _configuredMaxRetryDelay = _configuration.MaxRetryDelay; - - _backOff = new ExponentialBackoffWithDecorrelation(_configuredInitialRetryDelay, - _configuredMaxRetryDelay); - - _httpClient = _configuration.HttpClient ?? CreateHttpClient(); - } - - /// - /// Shortcut for initializing an with only a stream URI - /// and no custom properties. - /// - /// the stream URI - /// if the URI is null - public EventSource(Uri uri) : this(Configuration.Builder(uri).Build()) {} - - #endregion - - #region Public Methods - - /// - public async Task StartAsync() - { - bool firstTime = true; - while (ReadyState != ReadyState.Shutdown) - { - if (!firstTime) - { - if (_connectionTimer.IsRunning) - { - if (_connectionTimer.Elapsed >= _configuration.BackoffResetThreshold) - { - // Sustained healthy operation reverts any temporary bounds and resets - // n. Reverting takes precedence over bounds that were - // set during the preceding fault window. - ClearTemporaryRetryDelayBounds(); - _backOff.ResetBackoffN(); - } - _connectionTimer.Reset(); - } - await MaybeWaitWithBackOff(); - } - firstTime = false; - - CancellationTokenSource newRequestTokenSource = null; - CancellationToken cancellationToken; - lock (this) - { - if (_readyState == ReadyState.Shutdown) - { - // in case Close() was called in between the previous ReadyState check and the creation of the new token - return; - } - newRequestTokenSource = new CancellationTokenSource(); - _currentRequestToken?.Dispose(); - _currentRequestToken = newRequestTokenSource; - } - - if (ReadyState == ReadyState.Connecting || ReadyState == ReadyState.Open) - { - throw new InvalidOperationException(string.Format(Resources.ErrorAlreadyStarted, ReadyState)); - } - - SetReadyState(ReadyState.Connecting); - cancellationToken = newRequestTokenSource.Token; - - try - { - await ConnectToEventSourceAsync(cancellationToken); - - // ConnectToEventSourceAsync normally doesn't return, unless it detects that the request has been cancelled. - Close(ReadyState.Closed); - } - catch (Exception e) - { - CancelCurrentRequest(); - - // If the user called Close(), ReadyState = Shutdown, so errors are irrelevant. - if (ReadyState != ReadyState.Shutdown) - { - Exception realException = e; - if (realException is AggregateException ae && ae.InnerExceptions.Count == 1) - { - realException = ae.InnerException; - } - if (realException is OperationCanceledException oe) - { - // This exception could either be the result of us explicitly cancelling a request, in which case we don't - // need to do anything else, or it could be that the request timed out. - if (oe.CancellationToken.IsCancellationRequested) - { - realException = null; - } - else - { - realException = new TimeoutException(); - } - } - - if (realException != null) - { - OnError(new ExceptionEventArgs(realException)); - } - Close(ReadyState.Closed); - } - } - } - } - - private async Task MaybeWaitWithBackOff() { - TimeSpan sleepTime = _backOff.GetNextBackOff(); - if (sleepTime > MaxSleepTime) - { - // Task.Delay throws for anything above the platform timer ceiling, and this method - // runs outside the reconnect loop's exception handling, so an unclamped value - // would fault StartAsync and stop the stream permanently. - sleepTime = MaxSleepTime; - } - if (sleepTime > TimeSpan.Zero) - { - BackOffDelay = sleepTime; - try - { - await Task.Delay(sleepTime, _shutdownTokenSource.Token); - } - catch (OperationCanceledException) - { - // Cancellation happened during the wait, likely intentional close - } - } - } - - /// - public void Restart(bool resetBackoffDelay) - { - lock (this) - { - if (_readyState != ReadyState.Open) - { - return; - } - if (resetBackoffDelay) - { - _backOff.ResetBackoffN(); - } - } - CancelCurrentRequest(); - } - - /// - public void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay) - { - _backOff.SetBounds(initialDelay, maxDelay); - } - - /// - public void ClearTemporaryRetryDelayBounds() - { - _backOff.SetBounds(_configuredInitialRetryDelay, _configuredMaxRetryDelay); - } - - /// - /// Closes the connection to the SSE server. The EventSource cannot be reopened after this. - /// - public void Close() - { - if (ReadyState != ReadyState.Shutdown) - { - Close(ReadyState.Shutdown); - } - CancelCurrentRequest(); - _shutdownTokenSource.Cancel(); - - // do not dispose httpClient if it is user provided - if (_configuration.HttpClient == null) - { - _httpClient.Dispose(); - } - } - - /// - /// Equivalent to calling . - /// - public void Dispose() - { - Dispose(true); - } - - private void Dispose(bool disposing) - { - if (disposing) - { - Close(); - } - } - - #endregion - - #region Private Methods - - private HttpClient CreateHttpClient() - { - var client =_configuration.HttpMessageHandler is null ? - new HttpClient() : - new HttpClient(_configuration.HttpMessageHandler, false); - client.Timeout = _configuration.ResponseStartTimeout; - return client; - } - - private void CancelCurrentRequest() - { - CancellationTokenSource requestTokenSource = null; - lock (this) - { - requestTokenSource = _currentRequestToken; - _currentRequestToken = null; - } - if (requestTokenSource != null) - { - requestTokenSource.Cancel(); - requestTokenSource.Dispose(); - } - } - - internal virtual EventSourceService GetEventSourceService(Configuration configuration) - { - return new EventSourceService(configuration, _httpClient); - } - - private async Task ConnectToEventSourceAsync(CancellationToken cancellationToken) - { - _eventDataStringBuffer = null; - _eventDataUtf8ByteBuffer = null; - - var svc = GetEventSourceService(_configuration); - - svc.ConnectionOpened += (o, e) => { - _connectionTimer.Restart(); - SetReadyState(ReadyState.Open, OnOpened, e.Headers); - }; - svc.ConnectionClosed += (o, e) => { SetReadyState(ReadyState.Closed, OnClosed); }; - - await svc.GetDataAsync( - ProcessResponseLineString, - ProcessResponseLineUtf8, - _lastEventId, - cancellationToken - ); - } - - private void Close(ReadyState state) - { - SetReadyState(state, OnClosed); - } - - private void ProcessResponseLineString(string content) - { - if (content == null) - { - // StreamReader may emit a null if the stream has been closed; there's nothing to - // be done at this level in that case - return; - } - if (content.Length == 0) - { - DispatchEvent(); - } - else - { - HandleParsedLine(EventParser.ParseLineString(content)); - } - } - - private void ProcessResponseLineUtf8(Utf8ByteSpan content) - { - if (content.Length == 0) - { - DispatchEvent(); - } - else - { - HandleParsedLine(EventParser.ParseLineUtf8Bytes(content)); - } - } - - - private void SetReadyState(ReadyState state, Action action = null, - IEnumerable>> headers = null) - { - lock (this) - { - if (_readyState == state || _readyState == ReadyState.Shutdown) - { - return; - } - - _readyState = state; - } - - if (action != null) - { - action(new StateChangedEventArgs(state, headers)); - } - } - - private void HandleParsedLine(EventParser.Result result) - { - if (result.IsComment) - { - OnCommentReceived(new CommentReceivedEventArgs(result.GetValueAsString())); - } - else if (result.IsDataField) - { - if (result.ValueString != null) - { - if (_eventDataStringBuffer is null) - { - _eventDataStringBuffer = new List(2); - } - _eventDataStringBuffer.Add(result.ValueString); - _eventDataStringBuffer.Add("\n"); - } - else - { - if (_eventDataUtf8ByteBuffer is null) - { - _eventDataUtf8ByteBuffer = new MemoryStream(result.ValueBytes.Length + 1); - } - _eventDataUtf8ByteBuffer.Write(result.ValueBytes.Data, result.ValueBytes.Offset, result.ValueBytes.Length); - _eventDataUtf8ByteBuffer.WriteByte((byte)'\n'); - } - } - else if (result.IsEventField) - { - _eventName = result.GetValueAsString(); - } - else if (result.IsIdField) - { - _lastEventId = result.GetValueAsString(); - } - else if (result.IsRetryField) - { - if (long.TryParse(result.GetValueAsString(), out var retry)) - { - // Clamp while the value is still an integer. - var millis = Math.Max(0, - Math.Min(retry, Constants.MaxServerDirectedRetryDelayMillis)); - _backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(millis)); - } - } - } - - private void DispatchEvent() - { - var name = _eventName ?? Constants.MessageField; - _eventName = null; - MessageEvent message; - - if (_eventDataStringBuffer != null) - { - if (_eventDataStringBuffer.Count == 0) - { - return; - } - // remove last item which is always a trailing newline - _eventDataStringBuffer.RemoveAt(_eventDataStringBuffer.Count - 1); - var dataString = string.Concat(_eventDataStringBuffer); - message = new MessageEvent(name, dataString, _lastEventId, _configuration.Uri); - - _eventDataStringBuffer.Clear(); - } - else - { - if (_eventDataUtf8ByteBuffer is null || _eventDataUtf8ByteBuffer.Length == 0) - { - return; - } - var dataSpan = new Utf8ByteSpan(_eventDataUtf8ByteBuffer.GetBuffer(), 0, - (int)_eventDataUtf8ByteBuffer.Length - 1); // remove trailing newline - message = new MessageEvent(name, dataSpan, _lastEventId, _configuration.Uri); - - // We've now taken ownership of the original buffer; null out the previous - // reference to it so a new one will be created next time - _eventDataUtf8ByteBuffer = null; - } - - OnMessageReceived(new MessageReceivedEventArgs(message)); - } - - private void OnOpened(StateChangedEventArgs e) => - Opened?.Invoke(this, e); - - private void OnClosed(StateChangedEventArgs e) => - Closed?.Invoke(this, e); - - private void OnMessageReceived(MessageReceivedEventArgs e) => - MessageReceived?.Invoke(this, e); - - private void OnCommentReceived(CommentReceivedEventArgs e) => - CommentReceived?.Invoke(this, e); - - private void OnError(ExceptionEventArgs e) => - Error?.Invoke(this, e); - - #endregion - } + /// + /// Provides an EventSource client for consuming Server-Sent Events. Additional details on the Server-Sent Events spec + /// can be found at https://html.spec.whatwg.org/multipage/server-sent-events.html + /// + public class EventSource : IEventSource, IDisposable + { + #region Private Fields + + private readonly Configuration _configuration; + private readonly HttpClient _httpClient; + private readonly ILogger _logger; + + private List _eventDataStringBuffer; + private MemoryStream _eventDataUtf8ByteBuffer; + private string _eventName; + private string _lastEventId; + private readonly ExponentialBackoffWithDecorrelation _backOff; + private readonly TimeSpan _configuredInitialRetryDelay; + private readonly TimeSpan _configuredMaxRetryDelay; + private CancellationTokenSource _currentRequestToken; + private readonly CancellationTokenSource _shutdownTokenSource = new CancellationTokenSource(); + private static readonly TimeSpan MaxSleepTime = TimeSpan.FromMilliseconds(int.MaxValue); + private readonly Stopwatch _connectionTimer = new Stopwatch(); + private ReadyState _readyState; + + #endregion + + #region Public Events + + /// + public event EventHandler Opened; + + /// + public event EventHandler Closed; + + /// + public event EventHandler MessageReceived; + + /// + public event EventHandler CommentReceived; + + /// + public event EventHandler Error; + + #endregion Public Events + + #region Public Properties + + /// + public ReadyState ReadyState + { + get + { + lock (this) + { + return _readyState; + } + } + private set + { + lock (this) + { + _readyState = value; + } + } + } + + internal TimeSpan BackOffDelay + { + get; + private set; + } + + // Exposed for tests, so that retry-bounds behavior can be asserted directly + internal ExponentialBackoffWithDecorrelation BackOff => _backOff; + + #endregion + + #region Public Constructors + + /// + /// Initializes a new instance of the class. + /// + /// the configuration + public EventSource(Configuration configuration) + { + _readyState = ReadyState.Raw; + + _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); + + _logger = _configuration.Logger; + + _configuredInitialRetryDelay = _configuration.InitialRetryDelay; + _configuredMaxRetryDelay = _configuration.MaxRetryDelay; + + _backOff = new ExponentialBackoffWithDecorrelation(_configuredInitialRetryDelay, + _configuredMaxRetryDelay); + + _httpClient = _configuration.HttpClient ?? CreateHttpClient(); + } + + /// + /// Shortcut for initializing an with only a stream URI + /// and no custom properties. + /// + /// the stream URI + /// if the URI is null + public EventSource(Uri uri) : this(Configuration.Builder(uri).Build()) { } + + #endregion + + #region Public Methods + + /// + public async Task StartAsync() + { + bool firstTime = true; + while (ReadyState != ReadyState.Shutdown) + { + if (!firstTime) + { + if (_connectionTimer.IsRunning) + { + if (_connectionTimer.Elapsed >= _configuration.BackoffResetThreshold) + { + // Sustained healthy operation reverts any temporary bounds and resets + // n. Reverting takes precedence over bounds that were + // set during the preceding fault window. + ClearTemporaryRetryDelayBounds(); + _backOff.ResetBackoffN(); + } + _connectionTimer.Reset(); + } + await MaybeWaitWithBackOff(); + } + firstTime = false; + + CancellationTokenSource newRequestTokenSource = null; + CancellationToken cancellationToken; + lock (this) + { + if (_readyState == ReadyState.Shutdown) + { + // in case Close() was called in between the previous ReadyState check and the creation of the new token + return; + } + newRequestTokenSource = new CancellationTokenSource(); + _currentRequestToken?.Dispose(); + _currentRequestToken = newRequestTokenSource; + } + + if (ReadyState == ReadyState.Connecting || ReadyState == ReadyState.Open) + { + throw new InvalidOperationException(string.Format(Resources.ErrorAlreadyStarted, ReadyState)); + } + + SetReadyState(ReadyState.Connecting); + cancellationToken = newRequestTokenSource.Token; + + try + { + await ConnectToEventSourceAsync(cancellationToken); + + // ConnectToEventSourceAsync normally doesn't return, unless it detects that the request has been cancelled. + Close(ReadyState.Closed); + } + catch (Exception e) + { + CancelCurrentRequest(); + + // If the user called Close(), ReadyState = Shutdown, so errors are irrelevant. + if (ReadyState != ReadyState.Shutdown) + { + Exception realException = e; + if (realException is AggregateException ae && ae.InnerExceptions.Count == 1) + { + realException = ae.InnerException; + } + if (realException is OperationCanceledException oe) + { + // This exception could either be the result of us explicitly cancelling a request, in which case we don't + // need to do anything else, or it could be that the request timed out. + if (oe.CancellationToken.IsCancellationRequested) + { + realException = null; + } + else + { + realException = new TimeoutException(); + } + } + + if (realException != null) + { + OnError(new ExceptionEventArgs(realException)); + } + Close(ReadyState.Closed); + } + } + } + } + + private async Task MaybeWaitWithBackOff() + { + TimeSpan sleepTime = _backOff.GetNextBackOff(); + if (sleepTime > MaxSleepTime) + { + // Task.Delay throws for anything above the platform timer ceiling, and this method + // runs outside the reconnect loop's exception handling, so an unclamped value + // would fault StartAsync and stop the stream permanently. + sleepTime = MaxSleepTime; + } + if (sleepTime > TimeSpan.Zero) + { + _logger.Info(string.Format("Waiting {0} milliseconds before reconnecting...", sleepTime.TotalMilliseconds)); + BackOffDelay = sleepTime; + try + { + await Task.Delay(sleepTime, _shutdownTokenSource.Token); + } + catch (OperationCanceledException) + { + // Cancellation happened during the wait, likely intentional close + _logger.Debug("Backoff wait interrupted by shutdown"); + } + } + } + + /// + public void Restart(bool resetBackoffDelay) + { + lock (this) + { + if (_readyState != ReadyState.Open) + { + return; + } + if (resetBackoffDelay) + { + _backOff.ResetBackoffN(); + } + } + CancelCurrentRequest(); + } + + /// + public void SetTemporaryRetryDelayBounds(TimeSpan initialDelay, TimeSpan maxDelay) + { + _backOff.SetBounds(initialDelay, maxDelay); + } + + /// + public void ClearTemporaryRetryDelayBounds() + { + _backOff.SetBounds(_configuredInitialRetryDelay, _configuredMaxRetryDelay); + } + + /// + /// Closes the connection to the SSE server. The EventSource cannot be reopened after this. + /// + public void Close() + { + if (ReadyState != ReadyState.Shutdown) + { + Close(ReadyState.Shutdown); + } + CancelCurrentRequest(); + _shutdownTokenSource.Cancel(); + + // do not dispose httpClient if it is user provided + if (_configuration.HttpClient == null) + { + _httpClient.Dispose(); + } + } + + /// + /// Equivalent to calling . + /// + public void Dispose() + { + Dispose(true); + } + + private void Dispose(bool disposing) + { + if (disposing) + { + Close(); + } + } + + #endregion + + #region Private Methods + + private HttpClient CreateHttpClient() + { + var client = _configuration.HttpMessageHandler is null ? + new HttpClient() : + new HttpClient(_configuration.HttpMessageHandler, false); + client.Timeout = _configuration.ResponseStartTimeout; + return client; + } + + private void CancelCurrentRequest() + { + CancellationTokenSource requestTokenSource = null; + lock (this) + { + requestTokenSource = _currentRequestToken; + _currentRequestToken = null; + } + if (requestTokenSource != null) + { + _logger.Debug("Cancelling current request"); + requestTokenSource.Cancel(); + requestTokenSource.Dispose(); + } + } + + internal virtual EventSourceService GetEventSourceService(Configuration configuration) + { + return new EventSourceService(configuration, _httpClient, _logger); + } + + private async Task ConnectToEventSourceAsync(CancellationToken cancellationToken) + { + _eventDataStringBuffer = null; + _eventDataUtf8ByteBuffer = null; + + var svc = GetEventSourceService(_configuration); + + svc.ConnectionOpened += (o, e) => + { + _connectionTimer.Restart(); + SetReadyState(ReadyState.Open, OnOpened, e.Headers); + }; + svc.ConnectionClosed += (o, e) => { SetReadyState(ReadyState.Closed, OnClosed); }; + + await svc.GetDataAsync( + ProcessResponseLineString, + ProcessResponseLineUtf8, + _lastEventId, + cancellationToken + ); + } + + private void Close(ReadyState state) + { + _logger.Debug(string.Format("Close({0}) - state was {1}", state, ReadyState)); + SetReadyState(state, OnClosed); + } + + private void ProcessResponseLineString(string content) + { + if (content == null) + { + // StreamReader may emit a null if the stream has been closed; there's nothing to + // be done at this level in that case + return; + } + if (content.Length == 0) + { + DispatchEvent(); + } + else + { + HandleParsedLine(EventParser.ParseLineString(content)); + } + } + + private void ProcessResponseLineUtf8(Utf8ByteSpan content) + { + if (content.Length == 0) + { + DispatchEvent(); + } + else + { + HandleParsedLine(EventParser.ParseLineUtf8Bytes(content)); + } + } + + + private void SetReadyState(ReadyState state, Action action = null, + IEnumerable>> headers = null) + { + lock (this) + { + if (_readyState == state || _readyState == ReadyState.Shutdown) + { + return; + } + + _readyState = state; + } + + if (action != null) + { + action(new StateChangedEventArgs(state, headers)); + } + } + + private void HandleParsedLine(EventParser.Result result) + { + if (result.IsComment) + { + OnCommentReceived(new CommentReceivedEventArgs(result.GetValueAsString())); + } + else if (result.IsDataField) + { + if (result.ValueString != null) + { + if (_eventDataStringBuffer is null) + { + _eventDataStringBuffer = new List(2); + } + _eventDataStringBuffer.Add(result.ValueString); + _eventDataStringBuffer.Add("\n"); + } + else + { + if (_eventDataUtf8ByteBuffer is null) + { + _eventDataUtf8ByteBuffer = new MemoryStream(result.ValueBytes.Length + 1); + } + _eventDataUtf8ByteBuffer.Write(result.ValueBytes.Data, result.ValueBytes.Offset, result.ValueBytes.Length); + _eventDataUtf8ByteBuffer.WriteByte((byte)'\n'); + } + } + else if (result.IsEventField) + { + _eventName = result.GetValueAsString(); + } + else if (result.IsIdField) + { + _lastEventId = result.GetValueAsString(); + } + else if (result.IsRetryField) + { + if (long.TryParse(result.GetValueAsString(), out var retry)) + { + // Clamp while the value is still an integer. + var millis = Math.Max(0, + Math.Min(retry, Constants.MaxServerDirectedRetryDelayMillis)); + _backOff.SetServerDirectedMinDelay(TimeSpan.FromMilliseconds(millis)); + } + } + } + + private void DispatchEvent() + { + var name = _eventName ?? Constants.MessageField; + _eventName = null; + MessageEvent message; + + if (_eventDataStringBuffer != null) + { + if (_eventDataStringBuffer.Count == 0) + { + return; + } + // remove last item which is always a trailing newline + _eventDataStringBuffer.RemoveAt(_eventDataStringBuffer.Count - 1); + var dataString = string.Concat(_eventDataStringBuffer); + message = new MessageEvent(name, dataString, _lastEventId, _configuration.Uri); + + _eventDataStringBuffer.Clear(); + } + else + { + if (_eventDataUtf8ByteBuffer is null || _eventDataUtf8ByteBuffer.Length == 0) + { + return; + } + var dataSpan = new Utf8ByteSpan(_eventDataUtf8ByteBuffer.GetBuffer(), 0, + (int)_eventDataUtf8ByteBuffer.Length - 1); // remove trailing newline + message = new MessageEvent(name, dataSpan, _lastEventId, _configuration.Uri); + + // We've now taken ownership of the original buffer; null out the previous + // reference to it so a new one will be created next time + _eventDataUtf8ByteBuffer = null; + } + + _logger.Debug(string.Format("Received event \"{0}\"", name)); + OnMessageReceived(new MessageReceivedEventArgs(message)); + } + + private void OnOpened(StateChangedEventArgs e) => + Opened?.Invoke(this, e); + + private void OnClosed(StateChangedEventArgs e) => + Closed?.Invoke(this, e); + + private void OnMessageReceived(MessageReceivedEventArgs e) => + MessageReceived?.Invoke(this, e); + + private void OnCommentReceived(CommentReceivedEventArgs e) => + CommentReceived?.Invoke(this, e); + + private void OnError(ExceptionEventArgs e) => + Error?.Invoke(this, e); + + #endregion + } } diff --git a/OrderSupportTicket/LaunchDarkly.EventSource/EventSourceService.cs b/OrderSupportTicket/LaunchDarkly.EventSource/EventSourceService.cs index 4b21710..0bb5b94 100644 --- a/OrderSupportTicket/LaunchDarkly.EventSource/EventSourceService.cs +++ b/OrderSupportTicket/LaunchDarkly.EventSource/EventSourceService.cs @@ -7,6 +7,7 @@ using System.Net.Http.Headers; using System.Reflection; using System.Threading; using System.Threading.Tasks; +using WxOrder.Framework.Helpers; using static LaunchDarkly.EventSource.AsyncHelpers; @@ -23,6 +24,7 @@ namespace LaunchDarkly.EventSource private readonly Configuration _configuration; private readonly HttpClient _httpClient; + private readonly ILogger _logger; private const string UserAgentProduct = "DotNetClient"; internal static readonly string UserAgentVersion = ((AssemblyInformationalVersionAttribute)typeof(EventSource) @@ -48,10 +50,11 @@ namespace LaunchDarkly.EventSource #region Constructors - internal EventSourceService(Configuration configuration, HttpClient httpClient) + internal EventSourceService(Configuration configuration, HttpClient httpClient, ILogger logger) { _configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); _httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } #endregion @@ -85,16 +88,20 @@ namespace LaunchDarkly.EventSource CancellationToken cancellationToken ) { + _logger.Debug(string.Format("Making {0} request to EventSource URI {1}", + _configuration.Method ?? HttpMethod.Get, + _configuration.Uri)); var request = CreateHttpRequestMessage(_configuration.Uri, lastEventId); this._configuration.HttpRequestModifier?.Invoke(request); - using (var response = await _httpClient.SendAsync(request, + using (var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false)) { - ValidateResponse(response); + _logger.Debug(string.Format("Response status: {0}", (int)response.StatusCode)); + ValidateResponse(response); using (var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false)) { @@ -109,10 +116,12 @@ namespace LaunchDarkly.EventSource if (_configuration.PreferDataAsUtf8Bytes) { + _logger.Debug("Reading UTF-8 stream without string conversion"); await ProcessResponseFromUtf8StreamAsync(processResponseLineUTF8, stream, cancellationToken); } else { + _logger.Debug("Reading stream with string conversion"); using (var reader = new StreamReader(stream, Encoding.UTF8, detectEncodingFromByteOrderMarks: true, bufferSize: ReadBufferSize)) { diff --git a/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs b/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs index 93c960a..6e4e04a 100644 --- a/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs +++ b/OrderSupportTicket/Services/EventSourceTicketRealtimeService.cs @@ -5,6 +5,8 @@ using OrderSupportTicket.Models; using OrderSupportTicket.Models.Api; using System; using System.Linq; +using System.Net; +using System.Net.Http; using System.Threading.Tasks; using WxOrder.Framework.Helpers; @@ -31,7 +33,7 @@ public sealed class EventSourceTicketRealtimeService /// public TicketRealtimeState State { get; private set; } = TicketRealtimeState.Disconnected; - public void Start(TicketClientOptions options) + public async Task Start(TicketClientOptions options) { if (_disposed) throw new ObjectDisposedException(nameof(EventSourceTicketRealtimeService)); @@ -40,9 +42,8 @@ public sealed class EventSourceTicketRealtimeService TicketApiService.ValidateOptions(options); StopCore(); - - var uri = BuildEventsUri(options); - var source = new EventSource(uri); + var config = BuildEventsUri(options); + var source = new EventSource(config); source.Opened += OnOpened; source.Closed += OnClosed; source.MessageReceived += OnMessageReceived; @@ -55,7 +56,7 @@ public sealed class EventSourceTicketRealtimeService } SetState(TicketRealtimeState.Connecting, "连接中…"); - _ = RunAsync(source); + await RunAsync(source); } public Task StopAsync() @@ -77,7 +78,6 @@ public sealed class EventSourceTicketRealtimeService { try { - NLogger.Instance.Info("工单服务连接成"); await source.StartAsync(); lock (_sync) { @@ -182,22 +182,53 @@ public sealed class EventSourceTicketRealtimeService new TicketRealtimeStateChangedEventArgs(state, message)); } - private static Uri BuildEventsUri(TicketClientOptions options) + //private static Uri BuildEventsUri(TicketClientOptions options) + //{ + // var appIds = options.AppIds + // .Where(a => !string.IsNullOrWhiteSpace(a)) + // .Distinct(StringComparer.Ordinal) + // .Select(a => "appId=" + Uri.EscapeDataString(a.Trim())); + // var query = new[] + // { + // "purchaseAccount=" + Uri.EscapeDataString(options.UserName) + // }.Concat(appIds); + + // return new Uri( + // TicketApiService.ServerUrl + // + EventsPath + // + "?" + // + string.Join("&", query), + // UriKind.Absolute); + //} + + private static Configuration BuildEventsUri(TicketClientOptions options) { var appIds = options.AppIds .Where(a => !string.IsNullOrWhiteSpace(a)) .Distinct(StringComparer.Ordinal) - .Select(a => "appId=" + Uri.EscapeDataString(a.Trim())); + .Select(a => Uri.EscapeDataString(a.Trim())); var query = new[] { "purchaseAccount=" + Uri.EscapeDataString(options.UserName) - }.Concat(appIds); - - return new Uri( + }; + var uri = new Uri( TicketApiService.ServerUrl + EventsPath + "?" + string.Join("&", query), UriKind.Absolute); + ConfigurationBuilder builder = Configuration.Builder(uri); + builder.RequestHeader("X-AppIds", $"[\"{string.Join("\",\"", appIds)}\"]"); + builder.Logger(NLogger.Instance); + //var proxy = new WebProxy("http://127.0.0.1:8081"); // + //var handler = new HttpClientHandler + //{ + // Proxy = proxy, + // UseProxy = true + //}; + //var client = new HttpClient(handler); + //builder.HttpClient(client); + builder._responseStartTimeout = TimeSpan.FromSeconds(30); + return builder.Build(); } } diff --git a/OrderSupportTicket/Services/TicketApiService.cs b/OrderSupportTicket/Services/TicketApiService.cs index fbee627..2740135 100644 --- a/OrderSupportTicket/Services/TicketApiService.cs +++ b/OrderSupportTicket/Services/TicketApiService.cs @@ -372,8 +372,8 @@ public sealed class TicketApiService .ToList(); if (normalized.Count == 0) throw new ArgumentException("appIds 至少需要一个 APPID。", nameof(appIds)); - if (normalized.Count > 500) - throw new ArgumentException("appIds 最多支持 50 个 APPID。", nameof(appIds)); + //if (normalized.Count > 5000) + // throw new ArgumentException("appIds 最多支持 50 个 APPID。", nameof(appIds)); foreach (var appId in normalized) ValidateAppId(appId); return normalized; diff --git a/OrderSupportTicket/Services/TicketRealtimeMonitor.cs b/OrderSupportTicket/Services/TicketRealtimeMonitor.cs index bf3691d..aac9ff8 100644 --- a/OrderSupportTicket/Services/TicketRealtimeMonitor.cs +++ b/OrderSupportTicket/Services/TicketRealtimeMonitor.cs @@ -91,7 +91,7 @@ public sealed class TicketRealtimeMonitor : IDisposable /// /// 启动或复用实时连接。相同账号和 APPID 重复调用不会重建连接。 /// - public void Start(TicketClientOptions options) + public async Task Start(TicketClientOptions options) { if (_disposed) throw new ObjectDisposedException(nameof(TicketRealtimeMonitor)); @@ -113,7 +113,7 @@ public sealed class TicketRealtimeMonitor : IDisposable _optionsFingerprint = fingerprint; } - RealtimeService.Start(options); + await RealtimeService.Start(options); } /// diff --git a/OrderSupportTicket/Services/TicketService.cs b/OrderSupportTicket/Services/TicketService.cs index e063320..504eddf 100644 --- a/OrderSupportTicket/Services/TicketService.cs +++ b/OrderSupportTicket/Services/TicketService.cs @@ -167,11 +167,11 @@ public sealed class TicketService : IDisposable /// /// 使用指定工单参数建立 SSE 连接。 /// - public void StartRealtime(TicketClientOptions options) + public async Task StartRealtime(TicketClientOptions options) { ThrowIfDisposed(); ValidateOptions(options); - _realtime.Start(options); + await _realtime.Start(options); } /// @@ -191,7 +191,7 @@ public sealed class TicketService : IDisposable ThrowIfDisposed(); await _realtime.StopAsync(); ThrowIfDisposed(); - StartRealtime(options); + await StartRealtime(options); } public void Dispose() diff --git a/OrderSupportTicket/ViewModels/OrderSupportTicketViewModel.cs b/OrderSupportTicket/ViewModels/OrderSupportTicketViewModel.cs index 5df20d1..0953b06 100644 --- a/OrderSupportTicket/ViewModels/OrderSupportTicketViewModel.cs +++ b/OrderSupportTicket/ViewModels/OrderSupportTicketViewModel.cs @@ -714,7 +714,7 @@ public sealed class OrderSupportTicketViewModel : BindableBaseEx, IDialogAware, || (_ticketService.OwnsRealtime && !_initialized); if (shouldStartRealtime) { - _ticketService.StartRealtime(_options); + await _ticketService.StartRealtime(_options); started = true; } diff --git a/WxOrder.Framework/Helpers/ILogger.cs b/WxOrder.Framework/Helpers/ILogger.cs new file mode 100644 index 0000000..cb81701 --- /dev/null +++ b/WxOrder.Framework/Helpers/ILogger.cs @@ -0,0 +1,13 @@ +namespace WxOrder.Framework.Helpers; + +public interface ILogger +{ + void Debug(string message, Exception exception = null); + void Dispose(); + void Error(string message, Exception exception = null); + void Fatal(string message, Exception exception = null); + void Info(string message, Exception exception = null); + void Log(string message, LogMessageLevel level, Exception exception = null); + void Trace(string message, Exception exception = null); + void Warn(string message, Exception exception = null); +} \ No newline at end of file diff --git a/WxOrder.Framework/Helpers/NLogger.cs b/WxOrder.Framework/Helpers/NLogger.cs index b26ea2e..65759e2 100644 --- a/WxOrder.Framework/Helpers/NLogger.cs +++ b/WxOrder.Framework/Helpers/NLogger.cs @@ -32,7 +32,7 @@ namespace WxOrder.Framework.Helpers /// 日志文件按日期分目录存放，并在跨天时自动切换： /// {logDirectory}/yyyy-MM/yyyy-MM-dd.log，例如 log/2022-12/2022-12-12.log /// - public sealed class NLogger : IDisposable + public sealed class NLogger : IDisposable, ILogger { private const string DefaultName = "Default"; @@ -77,9 +77,9 @@ namespace WxOrder.Framework.Helpers /// 日志名称（写入每行日志中，用于区分来源，不再决定文件名）。 public string Name { get; } - public bool IsDebugEnabled { get; set; } = true; + public bool IsDebugEnabled { get; set; } = false; - public bool IsDebugWriteLineEnabled { get; set; } = true; + public bool IsDebugWriteLineEnabled { get; set; } = false; public LogMessageLevel MinimumLevel { get; set; } = LogMessageLevel.Trace;