Как я могу включить обработчики событий в мое потоковое приложение с сигнализатором? - PullRequest
Новая
       51

Как я могу включить обработчики событий в мое потоковое приложение с сигнализатором?

0 голосов
/ 16 июня 2019

Я знаю основы c # и ядра asp.net, однако мне нужна помощь с обработчиками событий. Я пытаюсь создать основное приложение asp.net, которое транслирует твиты на внешний интерфейс. Я использую библиотеку tweetinvi для получения твитов и сигнализатора для потоковой передачи.

Я тестировал tweetinvi в консольном приложении, и он работал. Я использовал обработчик событий и получил поток твитов, отображаемый на консоли. Тот же подход не работал в основном приложении asp.net. Однако мне удалось передать случайно сгенерированные числа на передний конец. Так что мой js-код на внешнем интерфейсе и настройка потокового концентратора на внутреннем сервере работали нормально. 

namespace TwitterWebApp
{
    public class StreamHub : Hub
    {
        public ChannelReader<string> GetRandomNumber(int count, CancellationToken cancellationToken)
        {
            var channel = Channel.CreateUnbounded<string>();


            StartTwitterStream(channel.Writer, count, cancellationToken);
            //_ = StreamRandomNumbers(channel.Writer, 100, cancellationToken);

            return channel.Reader;
        }

        private async Task StreamRandomNumbers(ChannelWriter<string> channelWriter, int count, CancellationToken cancellationToken)
        {

            var randomNumber = new Random();

            for (var i = 0; i < count; i++)
            {
                cancellationToken.ThrowIfCancellationRequested();
                await channelWriter.WriteAsync(Convert.ToString(randomNumber.Next()), cancellationToken);
            }

            channelWriter.TryComplete();
        }

        private void StartTwitterStream(ChannelWriter<string> channelWriter, int count, CancellationToken cancellationToken)
        {
            Auth.SetUserCredentials("xxx", "xxx", "xxx", "xxx");
            var stream = Stream.CreateSampleStream();
            stream.AddTweetLanguageFilter(LanguageFilter.English);

            stream.TweetReceived += async (sender, e) =>
            {
                var tweet = e.Tweet.Text;
                cancellationToken.ThrowIfCancellationRequested();
                await channelWriter.WriteAsync(tweet, cancellationToken);
            };

            stream.StartStream();
        }
    }
}

Получение твитов требует реализации обработчиков событий для потоковой передачи каждого твита после его получения. В отладчике я вижу твит и его содержание. Однако строка: await channelWriter.WriteAsync (tweet, cancellationToken); не отправляет строки в интерфейс. Нет ошибок в интерфейсе внешнего интерфейса.

Я сделал поток переменной класса и проверил хеш-код, и он имеет то же значение. Похоже, ручейка не убрали и не закрыли.

В режиме отладки действие TweetReceived вызывается бесконечно и генерирует новые твиты, но метод WriteAsync ничего не делает.

Позже, в начале, когда канал создается, я добавил 

channel.Writer.WriteAsync("ok", cancellationToken);

Просто чтобы послать строку "ок". Однако этот метод не выполняется, когда 

stream.StartStream();

вызывается позже. Я нахожу странным, почему WriteAsync не работает, когда StartStream вызывается намного позже. Удаление StartStream делает WriteSync отлично работающим. 

public class StreamHub : Hub
    {
        public ISampleStream stream;
        public ChannelReader<string> GetRandomNumber(int count, CancellationToken cancellationToken)
        {
            var channel = Channel.CreateUnbounded<string>();
            channel.Writer.WriteAsync("ok", cancellationToken);

            Auth.SetUserCredentials("xxx", "xxx", "xxx", "xxx");
            stream = Stream.CreateSampleStream();
            stream.AddTweetLanguageFilter(LanguageFilter.English);

            StartTwitterStream(channel.Writer, cancellationToken);

            return channel.Reader;
        }

        private void StartTwitterStream(ChannelWriter<string> channelWriter, CancellationToken cancellationToken)
        {
            stream.TweetReceived += (sender, e) =>
            {
                var streamHashCode1 = sender.GetHashCode();
                var tweet = e.Tweet.Text;
                cancellationToken.ThrowIfCancellationRequested();
                channelWriter.WriteAsync(tweet, cancellationToken);
            };

            var streamHashCode2 = stream.GetHashCode();
            stream.StartStream();
        }
    }
...