/// <summary>
/// 使用 async void 方法
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    // 使用 async Task 可以通过返回值的 Task 示例，监控任务的状态
    Task t = AsyncTask();
    t.Wait();

    // 使用 async void 没有返回值，无法监控任务状态
    AsyncVoid();
    // 这里使用 Sleep 方法确保任务完成
    Thread.Sleep(TimeSpan.FromSeconds(3));

    // 根据 Task.IsFaulted 属性可以判断是否发生异常
    // 捕获的异常信息可以从 Task.Exception 中获取
    t = AsyncTaskWithErrors();
    while (!t.IsFaulted)
    {
        Thread.Sleep(TimeSpan.FromSeconds(1));
    }
    Console.WriteLine(t.Exception);

    // 该段代码虽然使用 try/catch 捕获异常，但是由于使用了 async void 方法，
    // 异常处理方法会被放置到当前的同步上下文中（即线程池的线程中）。
    // 线程池中未被处理的异常会终止整个进程。

    //try
    //{
    //    AsyncVoidWithErrors();
    //    Thread.Sleep(TimeSpan.FromSeconds(3));
    //}
    //catch (Exception ex)
    //{
    //    Console.WriteLine(ex);
    //}

    // Action 类型也是可以使用 async 关键字的；
    // 在 lambda 表达式中很容易忘记对异常的处理，而这会导致程序崩溃。
    int[] numbers = { 1, 2, 3, 4, 5 };
    Array.ForEach(numbers, async number =>
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        if (number == 3)
        {
            throw new Exception("Boom!");
        }
        Console.WriteLine(number);
    });

    Console.ReadLine();
}

static async Task AsyncTaskWithErrors()
{
    string result = await GetInfoAsync("AsyncTaskException", 2);
    Console.WriteLine(result);
}

static async void AsyncVoidWithErrors()
{
    string result = await GetInfoAsync("AsyncVoidException", 2);
    Console.WriteLine(result);
}

static async Task AsyncTask()
{
    string result = await GetInfoAsync("AsyncTask", 2);
    Console.WriteLine(result);
}

static async void AsyncVoid()
{
    string result = await GetInfoAsync("AsyncVoid", 2);
    Console.WriteLine(result);
}

static async Task<string> GetInfoAsync(string name, int seconds)
{
    await Task.Delay(TimeSpan.FromSeconds(seconds));
    if (name.Contains("Exception"))
    {
        throw new Exception($"Boom from {name}");
    }
    return $"Task {name} is running on a thrad id {Thread.CurrentThread.ManagedThreadId}. " +
        $"Is thread pool thread: {Thread.CurrentThread.IsThreadPoolThread}";
}

/// <summary>
/// 避免使用捕获的同步上下文
/// </summary>
/// <param name="args"></param>
[STAThread]
static void Main(string[] args)
{
    var app = new Application();
    var win = new Window();
    var panel = new StackPanel();
    var button = new Button();

    _label = new Label();
    _label.FontSize = 32;
    _label.Height = 200;

    button.Height = 100;
    button.FontSize = 32;
    button.Content = new TextBlock { Text = "开始异步操作" };
    button.Click += Click;

    panel.Children.Add(_label);
    panel.Children.Add(button);

    win.Content = panel;
    app.Run(win);

    Console.ReadLine();
}

private static Label _label;

static async void Click(object sender, EventArgs e)
{
    _label.Content = new TextBlock { Text = "计算中......" };
    TimeSpan resultWithContent = await Test();
    TimeSpan resultNoContent = await TestNoContent();
    //TimeSpan resultNoContent = await TestNoContent().ConfigureAwait(false);

    var sb = new StringBuilder();
    sb.AppendLine($"With the content: {resultWithContent}");
    sb.AppendLine($"Without the content: {resultNoContent}");
    sb.AppendLine($"Ratio: {resultWithContent.TotalMilliseconds / resultNoContent.TotalMilliseconds:0.00}");

    _label.Content = new TextBlock { Text = sb.ToString() };
}

static async Task<TimeSpan> Test()
{
    const int iterationsNumber = 100000;
    var sw = new Stopwatch();
    sw.Start();
    for (int i = 0; i < iterationsNumber; i++)
    {
        var t = Task.Run(() => { });
        await t;
    }
    sw.Stop();
    return sw.Elapsed;
}

static async Task<TimeSpan> TestNoContent()
{
    const int interationsNumber = 100000;
    var sw = new Stopwatch();
    sw.Start();
    for (int i = 0; i < interationsNumber; i++)
    {
        var t = Task.Run(() => { });
        // 将 continueOnCapturedContext 指定为 false，不使用捕获的同步上下文运行后续操作代码
        await t.ConfigureAwait(continueOnCapturedContext: false);
    }
    sw.Stop();
    return sw.Elapsed;
}

/// <summary>
/// 处理异步操作中的异常
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    Task t = AsynchronousProcessing();
    t.Wait();

    Console.ReadLine();
}

static async Task AsynchronousProcessing()
{
    Console.WriteLine("1. 单个异常");
    Console.WriteLine("   使用 try/catch 捕获单个异常");
    try
    {
        string result = await GetInfoAsync("任务 1", 2);
        Console.WriteLine(result);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Exception details: {ex}");
    }

    Console.WriteLine();
    Console.WriteLine("2. 多个异常");
    Console.WriteLine("   尝试使用 try/catch 捕获多个异常，但此种写法仅能捕获其中一个异常");
    
    Task<string> t1 = GetInfoAsync("任务 1", 3);
    Task<string> t2 = GetInfoAsync("任务 2", 2);
    try
    {
        string[] results = await Task.WhenAll(t1, t2);
        Console.WriteLine(results.Length);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Exception details: {ex}");
    }

    Console.WriteLine();
    Console.WriteLine("3. 使用 AggregateException 捕获多个异常");
    Console.WriteLine("   从 task 的 Exception 属性中获取多个异常的信息");
    
    t1 = GetInfoAsync("任务 1", 3);
    t2 = GetInfoAsync("任务 2", 2);
    Task<string[]> t3 = Task.WhenAll(t1, t2);
    try
    {
        string[] results = await t3;
        Console.WriteLine(results.Length);
    }
    catch
    {
        var ae = t3.Exception.Flatten();
        var exceptions = ae.InnerExceptions;
        Console.WriteLine($"Exceptions caght: {exceptions.Count}");
        foreach (var e in exceptions)
        {
            Console.WriteLine($"Exception details: {e}");
            Console.WriteLine();
        }
    }

    Console.WriteLine();
    Console.WriteLine("4. 在 catch 和 finally 块中使用 await");
    Console.WriteLine("   c#6.0中，允许在catch和finally块中使用await，在之前的版本中这种写法会报错");
    Console.WriteLine("   若将语言版本改为C#5.0编译，则会报如下错误：");
    Console.WriteLine("     CS1985 无法在 catch 子句中等待");
    Console.WriteLine("     CS1984  无法在 finally 子句体中等待");
    
    try
    {
        string result = await GetInfoAsync("任务 1", 2);
        Console.WriteLine(result);
    }
    catch (Exception ex)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.WriteLine($"Catch block with await: Exception details: {ex}");
    }
    finally
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        Console.Write("Finally block");
    }
}

static async Task<string> GetInfoAsync(string name, int seconds)
{
    await Task.Delay(TimeSpan.FromSeconds(seconds));
    throw new Exception($"Boom from {name}");
}

/// <summary>
/// 对并行执行的异步任务使用 await 操作符
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    Task t = AsynchronousProcessing();
    t.Wait();

    Console.ReadLine();
}

static async Task AsynchronousProcessing()
{
    // 定义了两个任务，分别运行 3 秒和 5 秒
    Task<string> t1 = GetInfoAsync("Task 1", 3);
    Task<string> t2 = GetInfoAsync("Task 2", 5);
    // 使用 Task.WhenALL 辅助方法创建一个任务，该任务只有在所有底层任务完成后才会运行
    string[] results = await Task.WhenAll(t1, t2);
    // 5 秒后获得观察结果
    foreach (string result in results)
    {
        Console.WriteLine(result);
    }
}

static async Task<string> GetInfoAsync(string name, int seconds)
{
    // Task.Delay 方法不阻塞线程，而是在幕后使用了一个计时器
    // 导致两个任务一般总是由同一个线程运行
    await Task.Delay(TimeSpan.FromSeconds(seconds));
    // 如果使用下面注释掉的代码模拟运行时间，则两个任务总是由不同的线程运行
    // 因为 Thread.Sleep 方法阻塞线程
    // await Task.Run(() => Thread.Sleep(TimeSpan.FromSeconds(seconds)));
    return $"Task {name} is running on a thrad id {Thread.CurrentThread.ManagedThreadId}. " +
        $"Is thread pool thread: {Thread.CurrentThread.IsThreadPoolThread}";
}

/// <summary>
/// 可用于自身或导航至 Frame 内部的空白页。
/// </summary>
public sealed partial class MainPage : Page
{
    public MainPage()
    {
        this.InitializeComponent();
        // 初始化计时器
        _timer = new DispatcherTimer();
        _ticks = 0;
    }

    private readonly DispatcherTimer _timer;
    private int _ticks;

    protected override void OnNavigatedTo(NavigationEventArgs e)
    {
        base.OnNavigatedTo(e);
        Grid.Children.Clear();

        var commonPanel = new StackPanel
        {
            Orientation = Orientation.Vertical,
            HorizontalAlignment = HorizontalAlignment.Center
        };

        var buttonPanel = new StackPanel
        {
            Orientation = Orientation.Horizontal,
            HorizontalAlignment = HorizontalAlignment.Center
        };

        var textBlock = new TextBlock
        {
            Text = "Sample timer application",
            FontSize = 32,
            HorizontalAlignment = HorizontalAlignment.Center,
            Margin = new Thickness(40)
        };

        var timerTextBlock = new TextBlock
        {
            Text = "0",
            FontSize = 32,
            HorizontalAlignment = HorizontalAlignment.Center,
            Margin = new Thickness(40)
        };

        var timerStateTextBlock = new TextBlock
        {
            Text = "Timer is enabled",
            FontSize = 32,
            HorizontalAlignment = HorizontalAlignment.Center,
            Margin = new Thickness(40)
        };

        var startButton = new Button { Content = "Start", FontSize = 32 };
        var stopButton = new Button { Content = "Stop", FontSize = 32 };

        buttonPanel.Children.Add(startButton);
        buttonPanel.Children.Add(stopButton);

        commonPanel.Children.Add(textBlock);
        commonPanel.Children.Add(timerTextBlock);
        commonPanel.Children.Add(timerStateTextBlock);
        commonPanel.Children.Add(buttonPanel);

        // 设置间隔时间为 1s
        _timer.Interval = TimeSpan.FromSeconds(1);
        // 设置每个计时器间隔事件
        _timer.Tick += (sender, eventArgs) =>
        {
            timerTextBlock.Text = _ticks.ToString();
            _ticks++;
        };
        // 启动计时器
        _timer.Start();

        startButton.Click += (sender, eventArgs) =>
        {
            timerTextBlock.Text = "0";
            // 启动计时器
            _timer.Start();
            // 重置计数
            _ticks = 1;
            timerStateTextBlock.Text = "Timer is enabled";
        };

        stopButton.Click += (sender, eventArgs) =>
        {
            // 停止计时器
            _timer.Stop();
            timerStateTextBlock.Text = "Timer is disabled.";
        };

        Grid.Children.Add(commonPanel);
    }
}

/// <summary>
/// 使用 TPL 数据流实现并行管道
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var t = ProcessAsynchronously();
    t.GetAwaiter().GetResult();
}

async static Task ProcessAsynchronously()
{
    var cts = new CancellationTokenSource();
    Random _rnd = new Random(DateTime.Now.Millisecond);

    Task.Run(() =>
    {
            if (Console.ReadKey().KeyChar == 'c')
            {
                cts.Cancel();
            }
    }, cts.Token);

    // BufferBlock：将元素传给流中的下一个块
    // BoundedCapacity：指定其容量，超过时不再接受新元素，直到一个现有元素被传递给下一个块
    var inputBlock = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 5, CancellationToken = cts.Token });

    // TransformBlock：用于数据转换步骤
    //   MaxDegreeOfParallelism：通过该选项指定最大工作者线程数
    // 将 int 转换为 decimal
    var convertToDecimalBlock = new TransformBlock<int, decimal>(n =>
    {
        decimal result = Convert.ToDecimal(n * 100);
        Console.WriteLine($"Decimal Converter sent {result} to the next stage on {Thread.CurrentThread.ManagedThreadId}");
        Thread.Sleep(TimeSpan.FromMilliseconds(_rnd.Next(200)));
        return result;
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, CancellationToken = cts.Token });

    // 将 decimal 转换为 string
    var stringifyBlock = new TransformBlock<decimal, string>(n =>
    {
        string result = $"--{n.ToString("C", CultureInfo.GetCultureInfo("en-us"))}--";
        Console.WriteLine($"String Formatter sent {result} to the next stage on {Thread.CurrentThread.ManagedThreadId}");
        Thread.Sleep(TimeSpan.FromMilliseconds(_rnd.Next(200)));
        return result;
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, CancellationToken = cts.Token });

    // ActionBlock:对每个传入的元素运行一个指定的操作
    var outputBlock = new ActionBlock<string>(s => {
        Console.WriteLine($"The final result is {s} on thread id {Thread.CurrentThread.ManagedThreadId}");
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4, CancellationToken = cts.Token });

    // 通过 LinkTo 方法将这些块连接到一起
    // PropagateCompletion = true : 当前步骤完成时，自动将结果和异常传播到下一个阶段
    inputBlock.LinkTo(convertToDecimalBlock, new DataflowLinkOptions { PropagateCompletion = true });
    convertToDecimalBlock.LinkTo(stringifyBlock, new DataflowLinkOptions { PropagateCompletion = true });
    stringifyBlock.LinkTo(outputBlock, new DataflowLinkOptions { PropagateCompletion = true });

    try
    {
        // 向块中添加项
        Parallel.For(0, 20, new ParallelOptions {
            MaxDegreeOfParallelism = 4, CancellationToken = cts.Token
        }, i => {
            Console.WriteLine($"added {i} to source data on thread id {Thread.CurrentThread.ManagedThreadId}");
            inputBlock.SendAsync(i).GetAwaiter().GetResult();
        });
        // 添加完成后需要调用 Complete 方法
        inputBlock.Complete();
        // 等待最后的块完成
        await outputBlock.Completion;
        Console.WriteLine("Press ENTER to exit.");
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Operation has been canceled! Press ENTER to exit.");
    }

    Console.ReadLine();
}

private const int CollectionsNumber = 4;
private const int Count = 5;

/// <summary>
/// 使用 BlockingCollection 实现并行管道
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var cts = new CancellationTokenSource();

    // 监视 c 键（按下 c 键取消执行）
    Task.Run(() =>
    {
        if (Console.ReadKey().KeyChar == 'c')
        {
            cts.Cancel();
        }
    }, cts.Token);

    // 创建 4 个集合，每个集合中有 5 个元素
    var sourceArrays = new BlockingCollection<int>[CollectionsNumber];
    for (int i = 0; i < sourceArrays.Length; i++)
    {
        sourceArrays[i] = new BlockingCollection<int>(Count);
    }

    // 第一个管道：将 int 型数据转换成 Decimal 型
    var convertToDecimal = new PipelineWorker<int, decimal>(
        sourceArrays,
        n => Convert.ToDecimal(n * 100),
        cts.Token,
        "Decimal Converter"
        );

    // 第二个管道：格式化 Decimal 数据为金额字符串
    var stringifyNumber = new PipelineWorker<decimal, string>(
        convertToDecimal.Output,
        s => $"--{s.ToString("C", CultureInfo.GetCultureInfo("en-us"))}--",
        cts.Token,
        "Console Formatter"
        );

    // 第三个管道：打印格式化后的结果
    var outputResultToConsole = new PipelineWorker<string, string>(
        stringifyNumber.Output,
        s => Console.WriteLine($"The final result is {s} on thread id {Thread.CurrentThread.ManagedThreadId}"),
        cts.Token,
        "Console Output"
        );

    try
    {
        Parallel.Invoke(
            // 初始化集合数据
            () => CreateInitialValues(sourceArrays, cts),
            // 将 int 型数据转换成 Decimal 型
            () => convertToDecimal.Run(),
            // 格式化 Decimal 数据为金额字符串
            () => stringifyNumber.Run(),
            // 打印格式化后的结果
            () => outputResultToConsole.Run()
            );
    }
    catch (AggregateException ae)
    {
        foreach (var ex in ae.InnerExceptions)
        {
            Console.WriteLine(ex.Message + ex.StackTrace);
        }
    }

    if (cts.Token.IsCancellationRequested)
    {
        Console.WriteLine("Operation has been canceled! Press ENTER to exit.");
    }
    else
    {
        Console.WriteLine("Press ENTER to exit.");
    }

    Console.ReadLine();
}

/// <summary>
/// 初始化集合数据
/// </summary>
/// <param name="sourceArrays"></param>
/// <param name="cts"></param>
static void CreateInitialValues(BlockingCollection<int>[] sourceArrays, CancellationTokenSource cts)
{
    Parallel.For(0, sourceArrays.Length * Count, (j, state) =>
    {
        if (cts.Token.IsCancellationRequested)
        {
            state.Stop();
        }
        int number = GetRandomNumber(j);
        int k = BlockingCollection<int>.TryAddToAny(sourceArrays, j);
        if (k >= 0)
        {
            Console.WriteLine($"added {j} to source data on thread id {Thread.CurrentThread.ManagedThreadId}");
            Thread.Sleep(TimeSpan.FromMilliseconds(number));
        }
    });

    foreach (var arr in sourceArrays)
    {
        arr.CompleteAdding();
    }
}

static int GetRandomNumber(int seed)
{
    return new Random(seed).Next(500);
}

/// <summary>
/// 管道类
/// </summary>
/// <typeparam name="TInput"></typeparam>
/// <typeparam name="TOutput"></typeparam>
class PipelineWorker<TInput, TOutput>
{
    Func<TInput, TOutput> _processor;
    Action<TInput> _outputProcessor;
    BlockingCollection<TInput>[] _input;
    CancellationToken _token;
    Random _rnd;

    public BlockingCollection<TOutput>[] Output { get; private set; }

    public string Name { get; private set; }

    public PipelineWorker(
        BlockingCollection<TInput>[] input,
        Func<TInput, TOutput> processor,
        CancellationToken token,
        string name)
    {
        _input = input;
        Output = new BlockingCollection<TOutput>[_input.Length];
        for (int i = 0; i < Output.Length; i++)
        {
            Output[i] = null == input[i] ? null : new BlockingCollection<TOutput>(Count);
        }

        _processor = processor;
        _token = token;
        Name = name;
        _rnd = new Random(DateTime.Now.Millisecond);
    }

    public PipelineWorker(
        BlockingCollection<TInput>[] input,
        Action<TInput> renderer,
        CancellationToken token,
        string name)
    {
        _input = input;
        _outputProcessor = renderer;
        _token = token;
        Name = name;
        Output = null;
        _rnd = new Random(DateTime.Now.Millisecond);
    }

    public void Run()
    {
        Console.WriteLine($"{Name} is running");
        while (!_input.All(bc => bc.IsCompleted) && !_token.IsCancellationRequested)
        {
            TInput receivedItem;
            // 尝试从集合中获取元素；如没有元素则会等待
            int i = BlockingCollection<TInput>.TryTakeFromAny(_input, out receivedItem, 50, _token);
            if (i >= 0)
            {
                if (Output != null)
                {
                    TOutput outputItem = _processor(receivedItem);
                    BlockingCollection<TOutput>.AddToAny(Output, outputItem);
                    Console.WriteLine($"{Name} sent {outputItem} to next, on thread id {Thread.CurrentThread.ManagedThreadId}");
                    Thread.Sleep(TimeSpan.FromMilliseconds(_rnd.Next(200)));
                }
                else
                {
                    _outputProcessor(receivedItem);
                }
            }
            else
            {
                Thread.Sleep(TimeSpan.FromMilliseconds(50));
            }
        }

        if (Output != null)
        {
            foreach (var bc in Output)
            {
                bc.CompleteAdding();
            }
        }
    }
}

/// <summary>
/// 实现惰性求值的共享状态
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var t = ProcessAsynchronously();
    t.Wait();

    Console.WriteLine("Press Enter to exit");
    Console.ReadLine();
}

static async Task ProcessAsynchronously()
{
    // 不安全的对象，构造方法会被调用了多次，
    // 并且不同的线程中值是不同的
    var unsafeState = new UnsafeState();
    Task[] tasks = new Task[4];

    for (int i = 0; i < 4; i++)
    {
        tasks[i] = Task.Run(() => Worker(unsafeState));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine("------------------------------");

    // 使用双重锁定模式
    var firstState = new DoubleCheckedLocking();
    for (int i = 0; i < 4; i++)
    {
        tasks[i] = Task.Run(() => Worker(firstState));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine("------------------------------");

    // 使用 LazyInitializer.EnsureInitialized 方法
    var secondState = new BCLDoubleChecked();
    for (int i = 0; i < 4; i++)
    {
        tasks[i] = Task.Run(() => Worker(secondState));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine("------------------------------");

    // 使用 Lazy<T>类型
    var lazy = new Lazy<ValueToAccess>(Compute);
    var thirdState = new LazyWrapper(lazy);
    for (int i = 0; i < 4; i++)
    {
        tasks[i] = Task.Run(() => Worker(thirdState));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine("------------------------------");

    // 使用 LazyInitializer.EnsureInitialized 方法的一个不使用锁的重载
    var fourthState = new BCLThreadSafeFactory();
    for (int i = 0; i < 4; i++)
    {
        tasks[i] = Task.Run(() => Worker(fourthState));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine("------------------------------");
}

private static void Worker(IHasValue state)
{
    Console.WriteLine($"Worker runs on thread id {Thread.CurrentThread.ManagedThreadId}");
    Console.WriteLine($"State value: {state.Value.Text}");
}

class UnsafeState : IHasValue
{
    private ValueToAccess _value;

    // 不安全的对象，构造方法 Compute 会被调用多次
    public ValueToAccess Value => _value ?? (_value = Compute());
}

class DoubleCheckedLocking : IHasValue
{
    private readonly object _syncRoot = new object();
    private volatile ValueToAccess _value;
    
    public ValueToAccess Value
    {
        get
        {
            // 使用锁及双重验证，确保构造方法 Compute 仅执行一次
            if (_value == null)
            {
                lock (_syncRoot)
                {
                    if (_value == null)
                    {
                        _value = Compute();
                    }
                }
            }
            return _value;
        }
    }
}

class BCLDoubleChecked : IHasValue
{
    private object _syncRoot = new object();
    private ValueToAccess _value;
    private bool _initialized;

    // 使用 LazyInitializer.EnsureInitialized 方法初始化
    // 该方法内部实现了双重锁定模式
    public ValueToAccess Value => LazyInitializer.EnsureInitialized(ref _value, ref _initialized, ref _syncRoot, Compute);
}

class BCLThreadSafeFactory : IHasValue
{
    private ValueToAccess _value;

    // 使用 LazyInitializer.EnsureInitialized 方法初始化
    // 这个构造函数的重载没有使用锁，会导致初始化方法 Compute 被执行多次，但是结果的对象仍然是线程安全的
    public ValueToAccess Value => LazyInitializer.EnsureInitialized(ref _value, Compute);
}

class LazyWrapper : IHasValue
{
    // 使用 Lazy<T>类型
    // 效果同使用 LazyInitializer 一样
    // 区别是 LazyInitializer 是静态类，不需要初始化
    private readonly Lazy<ValueToAccess> _value;

    public LazyWrapper(Lazy<ValueToAccess> value)
    {
        _value = value;
    }

    public ValueToAccess Value => _value.Value;
}

static ValueToAccess Compute()
{
    Console.WriteLine($"The value is being constructed on a thread id {Thread.CurrentThread.ManagedThreadId}");
    Thread.Sleep(TimeSpan.FromSeconds(1));
    return new ValueToAccess($"Constructed on thread id {Thread.CurrentThread.ManagedThreadId}");
}

interface IHasValue
{
    ValueToAccess Value { get; }
}

class ValueToAccess
{
    private readonly string _text;
    public ValueToAccess(string text)
    {
        _text = text;
    }
    public string Text => _text;
}

/// <summary>
/// 异步调用 WCF 服务
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    ServiceHost host = null;

    try
    {
        // 创建一个的主机（服务类型为 HelloWorldService；承载服务的地址为 http://localhost:1234/HelloWorld）
        host = new ServiceHost(typeof(HelloWorldService), new Uri(SERVICE_URL));
        var metadata = host.Description.Behaviors.Find<ServiceMetadataBehavior>() ?? new ServiceMetadataBehavior();
        // 允许通过 HTTPGET 获取元数据
        // （即可以通过浏览器访问 http://localhost:1234/HelloWorld?wsdl 或 http://localhost:1234/HelloWorld?singleWsdl 获取 xml 格式的元数据）
        metadata.HttpGetEnabled = true;
        // 指定正在使用的 WS-Policy 规范的版本。
        metadata.MetadataExporter.PolicyVersion = PolicyVersion.Policy15;
        host.Description.Behaviors.Add(metadata);

        host.AddServiceEndpoint(ServiceMetadataBehavior.MexContractName, MetadataExchangeBindings.CreateMexHttpBinding(), "mex");

        var endpoint = host.AddServiceEndpoint(typeof(IHelloWorldService), new BasicHttpBinding(), SERVICE_URL);

        host.Faulted += (sender, e) => Console.WriteLine("Error!");

        // 开启主机
        host.Open();

        Console.WriteLine($"Greeting service is running and listening on: {endpoint.Address} ({endpoint.Binding.Name})");

        var client = RunServiceClient();
        client.GetAwaiter().GetResult();
        
        Console.WriteLine("Press Enter to exit");
        Console.ReadLine();
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Error in catch block: {ex}");
    }
    finally
    {
        if (null != host)
        {
            if (host.State == CommunicationState.Faulted)
            {
                host.Abort();
            }
            else
            {
                host.Close();
            }
        }
    }
}

const string SERVICE_URL = "http://localhost:1234/HelloWorld";

/// <summary>
/// 模拟远程访问 WCF 服务
/// </summary>
/// <returns></returns>
static async Task RunServiceClient()
{
    var endpoint = new EndpointAddress(SERVICE_URL);
    // 创建服务通道
    var channel = ChannelFactory<IHelloWorldServiceClient>.CreateChannel(new BasicHttpBinding(), endpoint);

    // 通过通道异步访问 Greet 服务
    var greeting = await channel.GreetAsync("Eugene");
    Console.WriteLine(greeting);
}

/// <summary>
/// 服务端接口
/// </summary>
[ServiceContract(Namespace = "Packt", Name = "HelloWorldServiceContract")]
public interface IHelloWorldService
{
    [OperationContract]
    string Greet(string name);
}

/// <summary>
/// 客户端服务接口
/// </summary>
[ServiceContract(Namespace = "Packt", Name = "HelloWorldServiceContract")]
public interface IHelloWorldServiceClient
{
    [OperationContract]
    string Greet(string name);

    [OperationContract]
    Task<string> GreetAsync(string name);
}

/// <summary>
/// 服务端实现
/// </summary>
public class HelloWorldService : IHelloWorldService
{
    public string Greet(string name)
    {
        return $"Greetings, {name}";
    }
}

/// <summary>
/// 异步操作数据库
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    const string dataBaseName = "CustomDatabase";
    var t = ProcessAsynchronousIO(dataBaseName);
    t.GetAwaiter().GetResult();
    Console.WriteLine("Press Enter to exit");
    Console.ReadLine();
}

static async Task ProcessAsynchronousIO(string dbName)
{
    try
    {
        const string connectionString = @"Data Source=(LocalDB)\MSSQLLocalDB;Initial Catalog=master;Integrated Security=True";
        string outputFolder = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);

        string dbFileName = Path.Combine(outputFolder, $"{dbName}.mdf");
        string dbLogFileName = Path.Combine(outputFolder, $"{dbName}_log.ldf");
        string dbConnectionString = $"Data Source=(LocalDB)\\MSSQLLocalDB;AttachDBFileName={dbFileName};Integrated Security=True;";

        using (var connection = new SqlConnection(connectionString))
        {
            await connection.OpenAsync();

            if (File.Exists(dbFileName))
            {
                Console.WriteLine("Detaching the database...");
                // 分离数据库（sp_detach_db：该命令常用来删除数据库日志文件或者移动数据库文件）
                var detachCommand = new SqlCommand("sp_detach_db", connection);
                detachCommand.CommandType = System.Data.CommandType.StoredProcedure;
                detachCommand.Parameters.AddWithValue("@dbname", dbName);

                await detachCommand.ExecuteNonQueryAsync();

                Console.WriteLine("The database was detached successfully.");
                Console.WriteLine("Deleting the database");

                // 删除数据库日志文件
                if (File.Exists(dbLogFileName))
                {
                    File.Delete(dbLogFileName);
                }
                // 删除数据库文件
                File.Delete(dbFileName);

                Console.WriteLine("The database was deleted successfully");
            }

            Console.WriteLine("Creating the database");
            // 创建数据库
            string createCommand = $"CREATE DATABASE {dbName} ON (NAME = N'{dbName}', FILENAME = '{dbFileName}')";
            var cmd = new SqlCommand(createCommand, connection);

            await cmd.ExecuteNonQueryAsync();

            Console.WriteLine("The database was created successfully");
        }

        using (var connection = new SqlConnection(dbConnectionString))
        {
            // 异步连接数据库
            await connection.OpenAsync();
            var cmd = new SqlCommand("SELECT newid()", connection);
            var result = await cmd.ExecuteNonQueryAsync();

            // 创建表
            cmd = new SqlCommand(@"CREATE TABLE [dbo].[CustomTable]([ID] [int] IDENTITY(1,1) NOT NULL,
[NAME] [nvarchar](50) NOT NULL,
CONSTRAINT [PK_ID] PRIMARY KEY CLUSTERED ([ID] ASC) ON [PRIMARY]) ON [PRIMARY]", connection);
            await cmd.ExecuteNonQueryAsync();
            Console.WriteLine("Table was created successfully");

            // 插入数据
            cmd = new SqlCommand(@"INSERT INTO [dbo].[CustomTable] (Name) VALUES ('John');
INSERT INTO [dbo].[CustomTable] (Name) VALUES ('Peter');
INSERT INTO [dbo].[CustomTable] (Name) VALUES ('james');
INSERT INTO [dbo].[CustomTable] (Name) VALUES ('Eugene');", connection);
            await cmd.ExecuteNonQueryAsync();
            Console.WriteLine("Inserted data successfully");
            Console.WriteLine("Reading data from table...");

            // 查询上面插入的数据
            cmd = new SqlCommand(@"SELECT * FROM [dbo].[CustomTable]", connection);
            // 异步查询数据
            using (SqlDataReader reader = await cmd.ExecuteReaderAsync())
            {
                // 异步读取查询结果
                while (await reader.ReadAsync())
                {
                    var id = reader.GetFieldValue<int>(0);
                    var name = reader.GetFieldValue<string>(1);

                    Console.WriteLine("Table row: Id {0}, Name {1}", id, name);
                }
            }
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine("Error: {0}", ex.Message);
    }
}

/// <summary>
/// 编写一个异步的 HTTP 服务器和客户端
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var server = new AsyncHttpServer(1234);
    var t = Task.Run(() => server.Start());
    Console.WriteLine("Listening on port 1234. Opent http://localhost:1234 iin your browser.");
    Console.WriteLine("Trying to conect:");
    Console.WriteLine();

    GetResponseAsync("http://localhost:1234").GetAwaiter().GetResult();

    Console.WriteLine();
    Console.WriteLine("Press Enter to stop the server");
    Console.ReadLine();

    server.Stop().GetAwaiter().GetResult();

    Console.WriteLine("Press Enter to exit");
    Console.ReadLine();
}

static async Task GetResponseAsync(string url)
{
    using (var client = new HttpClient())
    {
        HttpResponseMessage responseMessage = await client.GetAsync(url);
        string responseHeaders = responseMessage.Headers.ToString();
        string response = await responseMessage.Content.ReadAsStringAsync();

        Console.WriteLine("Response headers:");
        Console.WriteLine(responseHeaders);
        Console.WriteLine("Response body:");
        Console.WriteLine(response);
    }
}

class AsyncHttpServer
{
    // 使用 HttpListener 实现一个简单的服务器
    readonly HttpListener _listener;
    const string RESPONSE_TEMPLATE = "<html><head><title>Test</title></head><body><h2>Testpage</h2><h4>Today is: {0}</h4></body></html>";

    public AsyncHttpServer(int portNumber)
    {
        _listener = new HttpListener();
        _listener.Prefixes.Add($"http://localhost:{portNumber}/");
    }
    
    public async Task Start()
    {
        // 启动服务器
        _listener.Start();
        while (true)
        {
            // 调用 GetContextAsync 会发生异步 I/O 操作
            var ctx = await _listener.GetContextAsync();
            // 接收到请求时继续下面的处理
            Console.WriteLine("Client connected...");
            // 返回一个简单的 HTML 页面
            var response = string.Format(RESPONSE_TEMPLATE, DateTime.Now);

            using (var sw = new StreamWriter(ctx.Response.OutputStream))
            {
                await sw.WriteAsync(response);
                await sw.FlushAsync();
            }
        }
    }

    public async Task Stop()
    {
        // 调用 Abort 方法丢弃所有连接并关闭服务器
        _listener.Abort();
    }
}

/// <summary>
/// 异步地使用文件
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var t = ProcessAsynchronousIO();
    t.GetAwaiter().GetResult();

    Console.ReadLine();
}

const int BUFFER_SIZE = 4096;

static async Task ProcessAsynchronousIO()
{
    // 使用 FileStream 创建文件
    using (var stream = new FileStream("test1.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.None, BUFFER_SIZE))
    {
        Console.WriteLine($"1. Uses I/O Threads: {stream.IsAsync}");

        byte[] buffer = Encoding.UTF8.GetBytes(CreateFileContent());
        // 将异步编程模型 API 转换成任务
        var writeTask = Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, buffer, 0, buffer.Length, null);
        await writeTask;
    }

    // 使用 FileStream 创建文件（提供了 FileOptions.Asynchronous 参数）
    // 只有提供了提供了 FileOptions.Asynchronous 选项，才能对 FileStream 类使用异步 IO
    using (var stream = new FileStream("test2.txt", FileMode.Create, FileAccess.ReadWrite, FileShare.None, BUFFER_SIZE, FileOptions.Asynchronous))
    {
        Console.WriteLine($"2. Uses I/O Threads: {stream.IsAsync}");

        byte[] buffer = Encoding.UTF8.GetBytes(CreateFileContent());
        // 将异步编程模型 API 转换成任务
        var writeTask = Task.Factory.FromAsync(stream.BeginWrite, stream.EndWrite, buffer, 0, buffer.Length, null);
        await writeTask;
    }

    // 使用 File.Create（提供了 FileOptions.Asynchronous 参数）和 StreamWriter 创建和写入文件
    using (var stream = File.Create("test3.txt", BUFFER_SIZE, FileOptions.Asynchronous))
    using (var sw = new StreamWriter(stream))
    {
        Console.WriteLine($"3. Uses I/O Threads: {stream.IsAsync}");
        // 异步写入流
        await sw.WriteAsync(CreateFileContent());
    }

    // 仅使用 StreamWriter 创建文件
    using (var sw = new StreamWriter("test4.txt", true))
    {
        Console.WriteLine($"4. Uses I/O Threads: {((FileStream)sw.BaseStream).IsAsync}");

        // 异步写入流（因为没有提供 FileOptions.Asynchronous 参数，Stream 其实并没有使用异步 I/O）
        await sw.WriteAsync(CreateFileContent());
    }

    Console.WriteLine("Starting parsing files in parallel");
    var readTasks = new Task<long>[4];
    for (int i = 0; i < 4; i++)
    {
        string fileName = $"test{i + 1}.txt";
        // 异步读取文件并 Sum
        readTasks[i] = SumFileContent(fileName);
    }
    // 等待所有异步 Task 完成，并获取返回值数组
    long[] sums = await Task.WhenAll(readTasks);
    Console.WriteLine($"Sum is all files: {sums.Sum()}");

    Console.WriteLine("Deleting files");
    Task[] deleteTasks = new Task[4];
    for (int i = 0; i < 4; i++)
    {
        string filename = $"test{i + 1}.txt";
        // 异步删除文件
        deleteTasks[i] = SimulateAsynchronousDelete(filename);
    }
    // 等待所有异步删除操作结束
    await Task.WhenAll(deleteTasks);
    Console.WriteLine("Deleting complete.");
}

/// <summary>
/// 异步删除文件
/// </summary>
/// <param name="filename"></param>
/// <returns></returns>
static Task SimulateAsynchronousDelete(string filename)
{
    // 使用 Task.Run 模拟异步删除
    return Task.Run(() => File.Delete(filename));
}

/// <summary>
/// 异步统计文件中随机数的合计值
/// </summary>
/// <param name="fileName"></param>
/// <returns></returns>
static async Task<long> SumFileContent(string fileName)
{
    // 使用 FileStream 和 StreamReader 异步读取文件
    using (var stream = new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.None, BUFFER_SIZE, FileOptions.Asynchronous))
    using (var sr = new StreamReader(stream))
    {
        long sum = 0;
        while (sr.Peek() > -1)
        {
            string line = await sr.ReadLineAsync();
            sum += long.Parse(line);
        }
        return sum;
    }
}

/// <summary>
/// 创建随机的文件内容
/// </summary>
/// <returns></returns>
static string CreateFileContent()
{
    var sb = new StringBuilder();
    for (int i = 0; i < 100000; i++)
    {
        sb.Append($"{ new Random(DateTime.Now.Millisecond).Next(0, 99999)}");
        sb.AppendLine();
    }
    return sb.ToString();
}

delegate string AsyncDelegate(string name);

/// <summary>
/// 使用 Rx 创建异步操作
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    IObservable<string> o = LongRunningOperationAsync("Task1");
    using (var sub = OutputToConsole(o))
    {
        Thread.Sleep(TimeSpan.FromSeconds(2));
    }
    Console.WriteLine("--------------------");

    Task<string> t = LongRunningOperationTaskAsync("Task2");
    // 使用 ToObservable 方法将 Task 转换为 Observable 方法
    using (var sub = OutputToConsole(t.ToObservable()))
    {
        Thread.Sleep(TimeSpan.FromSeconds(2));
    }
    Console.WriteLine("--------------------");
    
    // 将异步编程模块模式转换为 Observable 类
    AsyncDelegate asyncMethod = LongRunningOperation;
    Func<string, IObservable<string>> observableFactory = Observable.FromAsyncPattern<string, string>(asyncMethod.BeginInvoke, asyncMethod.EndInvoke);
    o = observableFactory("Task3");
    using (var sub = OutputToConsole(o))
    {
        Thread.Sleep(TimeSpan.FromSeconds(2));
    }
    Console.WriteLine("--------------------");

    // 对 Observable 操作使用 await
    o = observableFactory("Task4");
    AwaitOnObservable(o).Wait();
    Console.WriteLine("--------------------");

    // 把基于事件的异步模式直接转换为 Observable 类
    using (var timer = new System.Timers.Timer(1000))
    {
        var ot = Observable.FromEventPattern<ElapsedEventHandler, ElapsedEventArgs>(
            h => timer.Elapsed += h
            , h => timer.Elapsed -= h
        );
        timer.Start();

        using (var sub = OutputToConsole(ot))
        {
            Thread.Sleep(TimeSpan.FromSeconds(5));
        }
        Console.WriteLine("--------------------");
        timer.Stop();
    }

    Console.ReadLine();
}

static async Task<T> AwaitOnObservable<T>(IObservable<T> observable)
{
    T obj = await observable;
    Console.WriteLine($"{obj}");
    return obj;
}

static Task<string> LongRunningOperationTaskAsync(string name)
{
    return Task.Run(() => LongRunningOperation(name));
}

static IObservable<string> LongRunningOperationAsync(string name)
{
    // Observable.Start 与 TPL 中的 Task.Run 方法很相似。
    // 启动异步操作并返回同一个字符串结果，然后退出
    return Observable.Start(() => LongRunningOperation(name));
}

static string LongRunningOperation(string name)
{
    Thread.Sleep(TimeSpan.FromSeconds(1));
    return $"Task {name} is completed. Thread id {Thread.CurrentThread.ManagedThreadId}";
}

static IDisposable OutputToConsole(IObservable<EventPattern<ElapsedEventArgs>> sequence)
{
    return sequence.Subscribe(
        obj => Console.WriteLine($"{obj.EventArgs.SignalTime}")
        , ex => Console.WriteLine($"Error: {ex.Message}")
        , () => Console.WriteLine("Completed")
    );
}

static IDisposable OutputToConsole<T>(IObservable<T> sequence)
{
    return sequence.Subscribe(
        obj => Console.WriteLine($"{obj}")
        , ex => Console.WriteLine($"Error: {ex.Message}")
        , () => Console.WriteLine("Completed")
    );
}

/// <summary>
/// 对可观察的集合使用 LINQ 查询
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    // 计时器序列
    IObservable<long> sequence = Observable.Interval(TimeSpan.FromMilliseconds(50)).Take(21);
    // 偶数序列
    var evenNumbers = from n in sequence
                        where n % 2 == 0
                        select n;
    // 奇数序列
    var oddNumbers = from n in sequence
                    where n % 2 != 0
                    select n;
    // 连接序列
    var combine = from n in evenNumbers.Concat(oddNumbers)
                    select n;
    // 副作用序列（Do 函数）
    var nums = (from n in combine
                where n % 5 == 0
                select n)
                .Do(n => Console.WriteLine($"------Number {n} is processed in Do method"));

    // 以上测查询均是惰性的，只有订阅了其结果，查询才会运行
    using (var sub = OutputToConsole(sequence, 0))
    using (var sub2 = OutputToConsole(combine, 1))
    using (var sub3 = OutputToConsole(nums, 2))
    { 
        Console.WriteLine("Press Enter to finish the demo");
        Console.ReadLine();
    }
}

static IDisposable OutputToConsole<T>(IObservable<T> sequence, int innerLevel)
{
    string delimiter = innerLevel == 0
        ? String.Empty
        : new string('-', innerLevel * 3);

    return sequence.Subscribe(
        obj => Console.WriteLine($"{delimiter}{obj}")
        , ex => Console.WriteLine($"Error: {ex.Message}")
        , () => Console.WriteLine($"{delimiter}Completed")
    );
}

/// <summary>
/// 创建可观察的对象
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    Console.WriteLine("使用值创建 Observable 方法");
    Console.WriteLine("Observable.Return(0)");
    IObservable<int> o = Observable.Return(0);
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("不使用值创建 Observable 方法");
    Console.WriteLine("Observable.Empty<int>()");
    o = Observable.Empty<int>();
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("通过 Observable.Throw 触发 OnError 处理器");
    Console.WriteLine("Observable.Throw<int>( new Exception())");
    o = Observable.Throw<int>( new Exception());
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("使用 Observable.Repeat 创建无尽序列");
    Console.WriteLine("Observable.Repeat(42)");
    o = Observable.Repeat(42);
    using (var sub = OutputToConsole(o.Take(5))) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("使用 Observable.Range 创建一组值");
    Console.WriteLine("Observable.Range(0, 10)");
    o = Observable.Range(0, 10);
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("Observable.Create 方法支持很多的自定义场景");
    // Create 方法接受一个函数，该函数接受一个观察者实例，并且返回 IDisposable 对象来代表订阅者
    o = Observable.Create<int>(ob =>
    {
        for (int i = 0; i < 10; i++)
        {
            ob.OnNext(i);
        }
        return Disposable.Empty;
    });
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("Observable.Generate 是另一个创建自定义序列的方式");
    o = Observable.Generate(
        0 // install state 初始值
        , i => i < 5 // while this is true we continue the sequence 一个断言，用来决定是否需要生成更多元素或者完成序列
        , i => ++i // iteration 迭代逻辑
        , i => i * 2  // selecting result 选择器函数，允许我们定制化结果
    );
    using (var sub = OutputToConsole(o)) ;
    Console.WriteLine("--------------------");

    Console.WriteLine("Interval 会以 TimeSpan 间隔产生计时器标记事件");
    Console.WriteLine("Observable.Interval(TimeSpan.FromSeconds(1))");
    IObservable<long> ol = Observable.Interval(TimeSpan.FromSeconds(1));
    using (var sub = OutputToConsole(ol))
    {
        Thread.Sleep(TimeSpan.FromSeconds(3));
    }
    Console.WriteLine("--------------------");


    Console.WriteLine("Timer 指定了启动时间");
    Console.WriteLine("Observable.Timer(DateTimeOffset.Now.AddSeconds(2))");
    ol = Observable.Timer(DateTimeOffset.Now.AddSeconds(2));
    using (var sub = OutputToConsole(ol))
    {
        Thread.Sleep(TimeSpan.FromSeconds(3));
    }
    Console.WriteLine("--------------------");

    Console.ReadLine();
}

static IDisposable OutputToConsole<T>(IObservable<T> sequence)
{
    return sequence.Subscribe(
        obj => Console.WriteLine($"{obj}")
        , ex => Console.WriteLine($"Error: {ex.Message}")
        , () => Console.WriteLine("Completed")
    );
}

/// <summary>
/// 使用 Subject
/// Subject 代表了 IObservable 和 IObserver 这两个接口的实现
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    // 一旦订阅了 Subject，它就会把事件序列发送给订阅者
    Console.WriteLine("Subject");
    var subject = new Subject<string>();
    
    subject.OnNext("A"); // A 在订阅之前，不会被打印
    using (var subscription = OutputToConsole(subject))
    {
        subject.OnNext("B");
        subject.OnNext("C");
        subject.OnNext("D");
        // 当调用 OnCompleted 或 OnError 方法时，事件序列传播会被停止
        subject.OnCompleted();
        // 事件传播停止之后的事件不会被打印
        subject.OnNext("Will not be printed out");
    }

    Console.WriteLine("ReplaySubject");
    // ReplaySubject 可以缓存从广播开始的所有事件
    var replaySubject = new ReplaySubject<string>();

    replaySubject.OnNext("A");
    // 稍后订阅也可以获得之前的事件
    using (var subscription = OutputToConsole(replaySubject))
    {
        replaySubject.OnNext("B");
        replaySubject.OnNext("C");
        replaySubject.OnNext("D");
        replaySubject.OnCompleted();
    }
    Console.WriteLine("Buffered ReplaySubject");
    // 指定 ReplaySubject 缓存的大小
    // 参数 2 表示只可以缓存最后的 2 个事件
    var bufferedSubject = new ReplaySubject<string>(2);

    bufferedSubject.OnNext("A");
    bufferedSubject.OnNext("B");
    bufferedSubject.OnNext("C");
    using (var subscription = OutputToConsole(bufferedSubject))
    {
        bufferedSubject.OnNext("D");
        bufferedSubject.OnCompleted();
    }

    Console.WriteLine("Time window ReplaySubject");
    // 指定 ReplaySubject 缓存的事件
    // TimeSpan.FromMilliseconds(200) 表示只缓存 200ms 内发生的事件
    var timeSubject = new ReplaySubject<string>(TimeSpan.FromMilliseconds(200));
    timeSubject.OnNext("A");
    Thread.Sleep(TimeSpan.FromMilliseconds(100));
    timeSubject.OnNext("B");
    Thread.Sleep(TimeSpan.FromMilliseconds(100));
    timeSubject.OnNext("C");
    Thread.Sleep(TimeSpan.FromMilliseconds(100));
    using (var subscription = OutputToConsole(timeSubject))
    {
        Thread.Sleep(TimeSpan.FromMilliseconds(300));
        timeSubject.OnNext("D");
        timeSubject.OnCompleted();
    }

    Console.WriteLine("AsyncSubject");
    // AsyncSubject 类似于任务并行库中的 Task 类型
    // 它代表了单个异步操作
    // 如果有多个事件发布，它将等待事件序列完成，并把最后一个事件提供给订阅者
    var asyncSubject = new AsyncSubject<string>();

    asyncSubject.OnNext("A");
    using (var subscription = OutputToConsole(asyncSubject))
    {
        asyncSubject.OnNext("B");
        asyncSubject.OnNext("C");
        asyncSubject.OnNext("D");
        asyncSubject.OnCompleted();
    }

    Console.WriteLine("BehaviorSubject");
    // BehaviorSubject 与 ReplaySubject 很相似，但它只缓存一个值
    // 并允许万一还没有发送任何通知时，指定一个默认值
    // 默认值会被自动替换为订阅前的最后一个事件
    var behaviorSubject = new BehaviorSubject<string>("Default");
    using (var subscription = OutputToConsole(behaviorSubject))
    {
        behaviorSubject.OnNext("B");
        behaviorSubject.OnNext("C");
        behaviorSubject.OnNext("D");
        behaviorSubject.OnCompleted();

    }

    Console.ReadLine();
}

static IDisposable OutputToConsole<T>(IObservable<T> sequence)
{
    return sequence.Subscribe(
        obj => Console.WriteLine($"{obj}")
        , ex => Console.WriteLine($"Error: {ex.Message}")
        , () => Console.WriteLine("Completed")
    );
}

/// <summary>
/// 编写自定义的可观察对象
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var observer = new CustomObserver();

    var goodObservable = new CustomSequence(new[] { 1, 2, 3, 4, 5 });
    var badObservable = new CustomSequence(null);

    // 同步订阅
    using (IDisposable subscription = goodObservable.Subscribe(observer))
    {
    }

    // 异步订阅
    using (IDisposable subscription = goodObservable.SubscribeOn(TaskPoolScheduler.Default).Subscribe(observer))
    {
        // 延迟一段时间等待异步任务完成
        Thread.Sleep(TimeSpan.FromMilliseconds(100));
        Console.WriteLine("Press Enter to continue");
        Console.ReadLine();
    }

    // 异步订阅异常时
    using (IDisposable subscription = badObservable.SubscribeOn(TaskPoolScheduler.Default).Subscribe(observer))
    {
        // 延迟一段时间等待异步任务完成
        Thread.Sleep(TimeSpan.FromMilliseconds(100));
        Console.WriteLine("Press Enter to continue");
        Console.ReadLine();
    }
}

/// <summary>
/// 自定义观察者
/// </summary>
class CustomObserver : IObserver<int>
{
    public void OnCompleted()
    {
        Console.WriteLine("Completed");
    }

    public void OnError(Exception error)
    {
        Console.WriteLine($"Error: {error.Message}");
    }

    public void OnNext(int value)
    {
        Console.WriteLine($"Next value: {value}; Thread id: {Thread.CurrentThread.ManagedThreadId}");
    }
}

/// <summary>
/// 自定义可观察对象
/// </summary>
class CustomSequence : IObservable<int>
{
    private readonly IEnumerable<int> _numbers;

    public CustomSequence(IEnumerable<int> numbers)
    {
        _numbers = numbers;
    }

    public IDisposable Subscribe(IObserver<int> observer)
    {
        foreach (var number in _numbers)
        {
            observer.OnNext(number);
        }
        observer.OnCompleted();
        return Disposable.Empty;
    }
}

/// <summary>
/// 管理 PLINQ 查询中的数据分区
/// </summary>
/// <param name="args"></param>
static void Main(string[] args)
{
    var sw = Stopwatch.StartNew();
    var partitioner = new StringPartitioner(GetTypes());
    var parallelQuery = from t in partitioner.AsParallel()
                        //.WithDegreeOfParallelism(1)
                        select EmulateProcessing(t);

    parallelQuery.ForAll(PrintInfo);
    int count = parallelQuery.Count();
    sw.Stop();
    Console.WriteLine("------------------------------");
    Console.WriteLine($"Total items processed: {count}");
    Console.WriteLine($"Time elapsesd: {sw.Elapsed}");

    Console.ReadLine();
}

static void PrintInfo(string typeName)
{
    Thread.Sleep(TimeSpan.FromMilliseconds(150));
    Console.WriteLine($"{typeName} type was printed on a thread id {Thread.CurrentThread.ManagedThreadId}");
}

static string EmulateProcessing(string typeName)
{
    Thread.Sleep(TimeSpan.FromMilliseconds(150));
    Console.WriteLine($"{typeName} type was processed on a thread id {Thread.CurrentThread.ManagedThreadId}. Has {(typeName.Length % 2 == 0 ? "even" : "odd")} length.");
    return typeName;
}

static IEnumerable<string> GetTypes()
{
    var types = AppDomain.CurrentDomain.GetAssemblies().SelectMany(a => a.GetExportedTypes());
    return from type in types
            where type.Name.StartsWith("Web")
            select type.Name;
}

/// <summary>
/// 自定义的分区器
/// </summary>
public class StringPartitioner : Partitioner<string>
{
    private readonly IEnumerable<string> _data;

    public StringPartitioner(IEnumerable<string> data)
    {
        _data = data;
    }

    /// <summary>
    /// 重写为 false，声明只支持静态分区
    /// </summary>
    public override bool SupportsDynamicPartitions => false;

    /// <summary>
    /// 重载生成静态分区方法（长度为奇数和偶数的字符串，分别放在不同的分区）
    /// </summary>
    /// <param name="partitionCount"></param>
    /// <returns></returns>
    public override IList<IEnumerator<string>> GetPartitions(int partitionCount)
    {
        var result = new List<IEnumerator<string>>(partitionCount);

        for (int i = 1; i <= partitionCount; i++)
        {
            result.Add(CreateEnumerator(i, partitionCount));
        }

        return result;
    }
    
    private IEnumerator<string> CreateEnumerator(int partitionNumber, int partitionCount)
    {
        int evenPartitions = partitionCount / 2;
        bool isEven = partitionNumber % 2 == 0;
        int step = isEven ? evenPartitions : partitionCount - evenPartitions;

        int startIndex = partitionNumber / 2 + partitionNumber % 2;

        var q = _data
            .Where(v => !(v.Length % 2 == 0 ^ isEven) || partitionCount == 1)
            .Skip(startIndex - 1);
        return q
            .Where((x, i) => i % step == 0)
            .GetEnumerator();
    }
}