<commit_before>﻿using System;
using NUnit.Framework;
using RethinkDb;
using System.Net;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Linq.Expressions;
using FluentAssertions;

namespace RethinkDb.Test.Integration
{
    [TestFixture]
    public class RealtimePushTests : TestBase
    {
        private ITableQuery<TestObject> testTable;

        public override void TestFixtureSetUp()
        {
            base.TestFixtureSetUp();
            connection.Run(Query.DbCreate("test"));
            connection.Run(Query.Db("test").TableCreate("table"));
            testTable = Query.Db("test").Table<TestObject>("table");
        }

        [SetUp]
        public virtual void SetUp()
        {
            connection.Run(testTable.Insert(new List<TestObject> {
                new TestObject() { Id = "1", Name = "1", SomeNumber = 1, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C1" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C1" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C1" } } },
                new TestObject() { Id = "2", Name = "2", SomeNumber = 2, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "3", Name = "3", SomeNumber = 3, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C3" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C3" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C3" } } },
                new TestObject() { Id = "4", Name = "4", SomeNumber = 4, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "5", Name = "5", SomeNumber = 5, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C5" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C5" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C5" } } },
                new TestObject() { Id = "6", Name = "6", SomeNumber = 6, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "7", Name = "7", SomeNumber = 7, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C7" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C7" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C7" } } },
            }));
        }

        [TearDown]
        public virtual void TearDown()
        {
            connection.Run(testTable.Delete());
        }

        //[Test]
        //public void ChangesOnPrimaryKey()
        //{
        //    foreach (var row in testTable.Get("Id").Changes())
        //    {
        //    }
        //}

        private void RealtimePushTestSingleResponse(
            Func<IStreamingSequenceQuery<DmlResponseChange<TestObject>>> createStreamingQuery,
            Action doModifications,
            Action<DmlResponseChange<TestObject>> verifyStreamingResults)
        {
            Exception e1 = null;
            Exception e2 = null;

            ManualResetEvent sync1 = new ManualResetEvent(false);

            var thread1 = new Thread(() =>
            {
                try
                {
                    var query = createStreamingQuery();
                    var enumerator = connection.StreamChangesAsync(query);
                    var task = enumerator.MoveNext();
                    sync1.Set(); // inform other thread that we're ready for it to make changes
                    task.Wait();
                    task.Result.Should().BeTrue();

                    verifyStreamingResults(enumerator.Current);
                }
                catch (Exception e)
                {
                    e1 = e;
                }
            });

            var thread2 = new Thread(() =>
            {
                try
                {
                    sync1.WaitOne();
                    doModifications();
                }
                catch (Exception e)
                {
                    e2 = e;
                }
            });

            thread1.Start();
            thread2.Start();

            thread1.Join();
            thread2.Join();

            e1.Should().BeNull();
            e2.Should().BeNull();
        }

        [Test]
        [Timeout(1000)]
        public void ChangesWithBetween()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Between("2", "4").Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { Name = "Updated!" }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Name.Should().Be("3");
                    response.NewValue.Name.Should().Be("Updated!");
                }
            );
        }
    }
}<commit_msg>Add tests for other realtime pushes that are supported<commit_after>﻿using System;
using NUnit.Framework;
using RethinkDb;
using System.Net;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Linq.Expressions;
using FluentAssertions;

namespace RethinkDb.Test.Integration
{
    [TestFixture]
    public class RealtimePushTests : TestBase
    {
        private ITableQuery<TestObject> testTable;

        public override void TestFixtureSetUp()
        {
            base.TestFixtureSetUp();
            connection.Run(Query.DbCreate("test"));
            connection.Run(Query.Db("test").TableCreate("table"));
            testTable = Query.Db("test").Table<TestObject>("table");
        }

        [SetUp]
        public virtual void SetUp()
        {
            connection.Run(testTable.Insert(new List<TestObject> {
                new TestObject() { Id = "1", Name = "1", SomeNumber = 1, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C1" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C1" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C1" } } },
                new TestObject() { Id = "2", Name = "2", SomeNumber = 2, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "3", Name = "3", SomeNumber = 3, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C3" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C3" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C3" } } },
                new TestObject() { Id = "4", Name = "4", SomeNumber = 4, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "5", Name = "5", SomeNumber = 5, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C5" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C5" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C5" } } },
                new TestObject() { Id = "6", Name = "6", SomeNumber = 6, Tags = new[] { "even" }, Children = new TestObject[0], ChildrenList = new List<TestObject> { }, ChildrenIList = new List<TestObject> { } },
                new TestObject() { Id = "7", Name = "7", SomeNumber = 7, Tags = new[] { "odd" }, Children = new TestObject[] { new TestObject { Name = "C7" } }, ChildrenList = new List<TestObject> { new TestObject() { Name = "C7" } }, ChildrenIList = new List<TestObject> { new TestObject() { Name = "C7" } } },
            }));
        }

        [TearDown]
        public virtual void TearDown()
        {
            connection.Run(testTable.Delete());
        }

        // Changes on a single record not currently supported by rethinkdb-net, but should be.
        //[Test]
        //public void ChangesOnPrimaryKey()
        //{
        //    foreach (var row in testTable.Get("Id").Changes())
        //    {
        //    }
        //}

        private void RealtimePushTestSingleResponse<T>(
            Func<IStreamingSequenceQuery<DmlResponseChange<T>>> createStreamingQuery,
            Action doModifications,
            Action<DmlResponseChange<T>> verifyStreamingResults)
        {
            Exception e1 = null;
            Exception e2 = null;

            ManualResetEvent sync1 = new ManualResetEvent(false);

            var thread1 = new Thread(() =>
            {
                try
                {
                    var query = createStreamingQuery();
                    var enumerator = connection.StreamChangesAsync(query);
                    var task = enumerator.MoveNext();
                    sync1.Set(); // inform other thread that we're ready for it to make changes
                    task.Wait();
                    task.Result.Should().BeTrue();

                    verifyStreamingResults(enumerator.Current);
                }
                catch (Exception e)
                {
                    e1 = e;
                }
            });

            var thread2 = new Thread(() =>
            {
                try
                {
                    sync1.WaitOne();
                    doModifications();
                }
                catch (Exception e)
                {
                    e2 = e;
                }
            });

            thread1.Start();
            thread2.Start();

            thread1.Join();
            thread2.Join();

            e1.Should().BeNull();
            e2.Should().BeNull();
        }

        [Test]
        [Timeout(1000)]
        public void ChangesWithBetween()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Between("2", "4").Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { Name = "Updated!" }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Name.Should().Be("3");
                    response.NewValue.Name.Should().Be("Updated!");
                }
            );
        }

        [Test]
        [Timeout(1000)]
        public void ChangesWithFilter()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Filter(o => o.Id == "3").Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { Name = "Updated!" }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Name.Should().Be("3");
                    response.NewValue.Name.Should().Be("Updated!");
                }
            );
        }

        [Test]
        [Timeout(1000)]
        public void ChangesWithMap()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Map(o => o.Name).Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { Name = "Updated!" }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Should().Be("3");
                    response.NewValue.Should().Be("Updated!");
                }
            );
        }

        [Test]
        [Timeout(1000)]
        [Ignore("Fails due to RethinkDB error 'cannot call changes on an eager stream' despite RethinkdB 1.16 documentation claiming this should work")]
        public void ChangesWithOrderByLimit()
        {
            RealtimePushTestSingleResponse(
                () => testTable.OrderBy(o => o.SomeNumber, OrderByDirection.Descending).Limit(1).Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { SomeNumber = 100 }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Id.Should().Be("7");
                    response.OldValue.SomeNumber.Should().Be(7);
                    response.NewValue.Id.Should().Be("3");
                    response.NewValue.SomeNumber.Should().Be(100);
                }
            );
        }

        /*
         * Min / Max operations on indexes not currently supported, but should be.

        [Test]
        [Timeout(1000)]
        public void ChangesWithMin()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Min(someNumberIndex).Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { SomeNumber = -100 }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Id.Should().Be("1");
                    response.OldValue.SomeNumber.Should().Be(1);
                    response.NewValue.Id.Should().Be("3");
                    response.NewValue.SomeNumber.Should().Be(-100);
                }
            );
        }

        [Test]
        [Timeout(1000)]
        public void ChangesWithMax()
        {
            RealtimePushTestSingleResponse(
                () => testTable.Max(someNumberIndex).Changes(),
                () =>
                {
                    var result = connection.Run(testTable.Get("3").Update(o => new TestObject() { SomeNumber = 100 }));
                    result.Should().NotBeNull();
                    result.Replaced.Should().Be(1.0);
                },
                response =>
                {
                    response.OldValue.Id.Should().Be("7");
                    response.OldValue.SomeNumber.Should().Be(7);
                    response.NewValue.Id.Should().Be("3");
                    response.NewValue.SomeNumber.Should().Be(100);
                }
            );
        }
        */
    }
}
{'fraction_non_alphanumeric': 0.135639721627409, 'fraction_numerical': 0.012781049250535331, 'mean_word_length': 1.5656652360515022, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3013107', 'n_tokens_mistral': 4077, 'n_tokens_neox': 3840, 'n_words': 839}