2.1.0 nuget package

Author: IgorBuchelnikov

src/ObservableComputations.Test/CollectionDispatchingTest.cs

@@ -152,23 +152,94 @@ public void TestPropertyDispatchingProperties()
 			backgroundOcDispatcher.Dispose();
 		}
 
+		byte _initWay = 0;
+
 		[Test]
-		[Repeat(1)]
+		[Repeat(10)]
 		[Timeout(1000 * 60 * 60 * 5)]
 		public void TestCollectionDispatchingTest()
 		{
+
 			OcDispatcher mainOcDispatcher = new OcDispatcher();
 			mainOcDispatcher.ThreadName = "mainOcDispatcher";
 			OcDispatcher backgroundOcDispatcher = new OcDispatcher();
 			backgroundOcDispatcher.ThreadName = "backgroundOcDispatcher";
 			OcConsumer consumer = new OcConsumer();
 
 			ObservableCollection<Item> nums = new ObservableCollection<Item>();
+
 			Filtering<Item> filteredNums = nums.Filtering(i =>
 				i.Num % 3 == 0
 				|| i.Num2MainToBackgroundDispatching.Value % 5 == 0
 				|| i.Num2MainToBackgroundScalarDispatching.Value % 5 == 0);
-			CollectionDispatching<Item> dispatchingfilteredNums = filteredNums
+
+			CollectionDispatching<Item> dispatchingfilteredNums;
+			ManualResetEvent manualResetEvent = new ManualResetEvent(false);
+
+			Action func = () =>
+			{
+				Thread.Sleep(TimeSpan.FromSeconds(new Random().Next(1, 4)));
+				dispatchingfilteredNums = filteredNums
+					.CollectionDispatching(mainOcDispatcher, backgroundOcDispatcher).For(consumer);
+
+				dispatchingfilteredNums.CollectionChanged += (sender, args) =>
+				{
+					if (Thread.CurrentThread != mainOcDispatcher._thread)
+					{
+						throw new Exception("Wrong thread");
+					}
+				};
+
+				((INotifyPropertyChanged)dispatchingfilteredNums).PropertyChanged += (sender, args) =>
+				{
+					if (Thread.CurrentThread != mainOcDispatcher._thread)
+					{
+						throw new Exception("Wrong thread");
+					}
+				};
+
+				filteredNums.CollectionChanged += (sender, args) =>
+				{
+
+					if (Thread.CurrentThread != backgroundOcDispatcher._thread)
+					{
+						throw new Exception("Wrong thread");
+					}
+				};
+
+				((INotifyPropertyChanged)filteredNums).PropertyChanged += (sender, args) =>
+				{
+					if (Thread.CurrentThread != backgroundOcDispatcher._thread)
+					{
+						throw new Exception("Wrong thread");
+					}
+				};
+
+				manualResetEvent.Set();
+			};
+
+			if (_initWay == 0)
+			{
+				mainOcDispatcher.InvokeAsync(func);
+			}
+			else if (_initWay == 1)
+			{
+				Thread thread = new Thread(() => func());
+				thread.Start();
+			}
+			else if (_initWay == 2)
+			{
+				func();
+			}
+			else 
+			{
+				backgroundOcDispatcher.InvokeAsync(func);
+			}
+
+			_initWay++;
+			if (_initWay == 4) _initWay = 0;
+
+			dispatchingfilteredNums = filteredNums
 				.CollectionDispatching(mainOcDispatcher, backgroundOcDispatcher).For(consumer);
 
 			dispatchingfilteredNums.CollectionChanged += (sender, args) =>
@@ -209,7 +280,7 @@ public void TestCollectionDispatchingTest()
 			Random stopperRandom = new Random();
 			Thread stopper = new Thread(() =>
 			{
-				Thread.Sleep(TimeSpan.FromSeconds(stopperRandom.Next(2, 20)));
+				Thread.Sleep(TimeSpan.FromSeconds(stopperRandom.Next(5, 20)));
 				stop = true;
 			});
 
@@ -309,6 +380,7 @@ public void TestCollectionDispatchingTest()
 
 			ThreadStart numValueChangerThreadStart = () =>
 			{
+				manualResetEvent.WaitOne();
 				Random random = new Random();
 				while (!stop)
 				{
@@ -335,6 +407,7 @@ public void TestCollectionDispatchingTest()
 
 			ThreadStart numValueChanger2ThreadStart = () =>
 			{
+				manualResetEvent.WaitOne();
 				Random random = new Random();
 				while (!stop)
 				{
@@ -362,6 +435,7 @@ public void TestCollectionDispatchingTest()
 
 			ThreadStart num2ValueChangerThreadStart = () =>
 			{
+				manualResetEvent.WaitOne();
 				Random random = new Random();
 				while (!stop)
 				{
@@ -405,6 +479,8 @@ public void TestCollectionDispatchingTest()
 				num2ValueChangerThreads[i].Join();
 			}
 
+			manualResetEvent.Dispose();
+
 			//consuminingOcDispatcherInvoker.Join();
 
 			mainOcDispatcher.Invoke(() => { });