Observable bucket with counts and a little graph

ObservableDistribution.linq

<Query Kind="Program">
  <Connection>
    <ID>0515683c-2a33-40ff-8c59-990ea709b2c8</ID>
    <Persist>true</Persist>
    <Server>rdbtest2012</Server>
    <Database>PackageMonitorReadModel_SGSpike-Main</Database>
    <ShowServer>true</ShowServer>
  </Connection>
  <Reference>&lt;RuntimeDirectory&gt;\System.Windows.Forms.DataVisualization.dll</Reference>
  <Reference>&lt;RuntimeDirectory&gt;\System.Windows.Forms.dll</Reference>
  <NuGetReference>Rx-Linq</NuGetReference>
  <NuGetReference>Rx-Main</NuGetReference>
  <NuGetReference>Rx-WPF</NuGetReference>
  <Namespace>System</Namespace>
  <Namespace>System.Drawing</Namespace>
  <Namespace>System.Net</Namespace>
  <Namespace>System.Reactive</Namespace>
  <Namespace>System.Reactive.Concurrency</Namespace>
  <Namespace>System.Reactive.Disposables</Namespace>
  <Namespace>System.Reactive.Joins</Namespace>
  <Namespace>System.Reactive.Linq</Namespace>
  <Namespace>System.Reactive.PlatformServices</Namespace>
  <Namespace>System.Reactive.Subjects</Namespace>
  <Namespace>System.Reactive.Threading.Tasks</Namespace>
  <Namespace>System.Threading.Tasks</Namespace>
  <Namespace>System.Windows.Forms</Namespace>
  <Namespace>System.Windows.Forms.DataVisualization.Charting</Namespace>
</Query>

void Main()
{
	var rand = new Random(5);
	var x =Observable.Range(0,200000)
	.Select(r=>r * rand.Next(3,7)).Publish();
	var y = x.GroupBy(r => r / 1024)
	.SelectMany(async rg => new Holder{ Key = (int)rg.Key, Count = await rg.Count().LastAsync()});
	x.Count().Dump("there how many?");
	var myl = new List<Holder>();
	var sub = y.Subscribe(Observer.Create<Holder>(gg=>myl.Add(gg)));
	x.Connect();
	myl.OrderBy(m => m.Key);//.Dump();

	var currentDate = DateTime.Parse("8/1/2016");
	var withDates = Observable.Range(0, 100)
		.Select(i =>
	{
		if ((i % 5) == 0)
		{
			currentDate = currentDate.AddDays(1);
		}
		return new { Value = i, Date = currentDate};
	});
	
	var chart = new Pie("My chart");

	var subscription = withDates.Buffer(withDates.Select(i => i.Date).DistinctUntilChanged())
	.Select(v => new PieSlice() { Text = v.FirstOrDefault().Date , Value = v.Sum(b => b.Value) })
	.Subscribe(Observer.Create<PieSlice>(v => chart.SetData(v)));
}

// Define other methods and classes here
public class Holder
{
	public int Key;
	public int Count = 0;
}

class Pie
{
	Chart chart = new Chart();
	Series series = new Series { ChartType = SeriesChartType.FastLine };

	public Pie(string name)
	{
		//series.CustomProperties = "PieDrawingStyle=Concave";
		chart.Series.Add(series);
		chart.ChartAreas.Add(new ChartArea() );
		chart.Dump(name);
	}

	public void SetData(IEnumerable<PieSlice> slices)
	{
		series.Points.SuspendUpdates();
		series.Points.Clear();
		foreach (var slice in slices) series.Points.AddXY(slice.Text, slice.Value);
		series.Points.ResumeUpdates();
	}

	public void SetData(PieSlice slice)
	{
		//series.Points.SuspendUpdates();
		series.Points.AddXY(slice.Text, slice.Value);
		//series.Points.ResumeUpdates();
	}
}

class PieSlice
{
	public DateTime Text;
	public decimal Value;
}