sam/BirdsiteLIVE
sam/BirdsiteLIVE
Archived
1
0
Fork 0
Code Issues 1 Pull requests 1 Projects Releases Packages 1 Wiki Activity

remove dead code

Browse source
This commit is contained in:
Vincent Cloutier 2023-01-01 15:18:54 -05:00
parent 676979150f
commit 3e5b01a923
8 changed files with 23 additions and 278 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/BirdsiteLive.Pipeline/Contracts/IRefreshTwitterUserStatusProcessor.cs
View file

@ -1,12 +0,0 @@
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
namespace BirdsiteLive.Pipeline.Contracts
{
public interface IRefreshTwitterUserStatusProcessor
{
Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct);
}
}

4
src/BirdsiteLive.Pipeline/Contracts/IRetrieveTwitterUsersProcessor.cs
View file

@ -1,12 +1,12 @@
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
namespace BirdsiteLive.Pipeline.Contracts
{
public interface IRetrieveTwitterUsersProcessor
{
Task GetTwitterUsersAsync(BufferBlock<SyncTwitterUser[]> twitterUsersBufferBlock, CancellationToken ct);
Task GetTwitterUsersAsync(BufferBlock<UserWithDataToSync[]> twitterUsersBufferBlock, CancellationToken ct);
}
}

109
src/BirdsiteLive.Pipeline/Processors/RefreshTwitterUserStatusProcessor.cs
View file

@ -1,109 +0,0 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.Common.Settings;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Moderation.Actions;
using BirdsiteLive.Pipeline.Contracts;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Twitter;
using BirdsiteLive.Twitter.Models;
namespace BirdsiteLive.Pipeline.Processors
{
public class RefreshTwitterUserStatusProcessor : IRefreshTwitterUserStatusProcessor
{
private readonly ICachedTwitterUserService _twitterUserService;
private readonly ITwitterUserDal _twitterUserDal;
private readonly IRemoveTwitterAccountAction _removeTwitterAccountAction;
private readonly InstanceSettings _instanceSettings;
#region Ctor
public RefreshTwitterUserStatusProcessor(ICachedTwitterUserService twitterUserService, ITwitterUserDal twitterUserDal, IRemoveTwitterAccountAction removeTwitterAccountAction, InstanceSettings instanceSettings)
{
_twitterUserService = twitterUserService;
_twitterUserDal = twitterUserDal;
_removeTwitterAccountAction = removeTwitterAccountAction;
_instanceSettings = instanceSettings;
}
#endregion
public async Task<UserWithDataToSync[]> ProcessAsync(SyncTwitterUser[] syncTwitterUsers, CancellationToken ct)
{
var usersWtData = new List<UserWithDataToSync>();
foreach (var user in syncTwitterUsers)
{
TwitterUser userView = null;
try
{
//userView = await _twitterUserService.GetUserAsync(user.Acct);
}
catch (UserNotFoundException)
{
await ProcessNotFoundUserAsync(user);
continue;
}
catch (UserHasBeenSuspendedException)
{
await ProcessNotFoundUserAsync(user);
continue;
}
catch (RateLimitExceededException)
{
await ProcessRateLimitExceededAsync(user);
continue;
}
catch (Exception)
{
// ignored
}
if (userView == null || userView.Protected)
{
//await ProcessFailingUserAsync(user);
//continue;
}
user.FetchingErrorCount = 0;
var userWtData = new UserWithDataToSync
{
User = user
};
usersWtData.Add(userWtData);
}
return usersWtData.ToArray();
}
private async Task ProcessRateLimitExceededAsync(SyncTwitterUser user)
{
var dbUser = await _twitterUserDal.GetTwitterUserAsync(user.Acct);
dbUser.LastSync = DateTime.UtcNow;
await _twitterUserDal.UpdateTwitterUserAsync(dbUser);
}
private async Task ProcessNotFoundUserAsync(SyncTwitterUser user)
{
await _removeTwitterAccountAction.ProcessAsync(user);
}
private async Task ProcessFailingUserAsync(SyncTwitterUser user)
{
var dbUser = await _twitterUserDal.GetTwitterUserAsync(user.Acct);
dbUser.FetchingErrorCount++;
dbUser.LastSync = DateTime.UtcNow;
if (dbUser.FetchingErrorCount > _instanceSettings.FailingTwitterUserCleanUpThreshold)
{
await _removeTwitterAccountAction.ProcessAsync(user);
}
else
{
await _twitterUserDal.UpdateTwitterUserAsync(dbUser);
}
}
}
}

7
src/BirdsiteLive.Pipeline/Processors/RetrieveTwitterUsersProcessor.cs
View file

@ -6,7 +6,7 @@ using System.Threading.Tasks.Dataflow;
using BirdsiteLive.Common.Extensions;
using BirdsiteLive.Common.Settings;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Contracts;
using BirdsiteLive.Pipeline.Tools;
using Microsoft.Extensions.Logging;
@ -30,7 +30,7 @@ namespace BirdsiteLive.Pipeline.Processors
}
#endregion
public async Task GetTwitterUsersAsync(BufferBlock<SyncTwitterUser[]> twitterUsersBufferBlock, CancellationToken ct)
public async Task GetTwitterUsersAsync(BufferBlock<UserWithDataToSync[]> twitterUsersBufferBlock, CancellationToken ct)
{
for (; ; )
{
@ -51,8 +51,9 @@ namespace BirdsiteLive.Pipeline.Processors
foreach (var u in splitUsers)
{
ct.ThrowIfCancellationRequested();
UserWithDataToSync[] toSync = u.Select(x => new UserWithDataToSync { User = x }).ToArray();
await twitterUsersBufferBlock.SendAsync(u.ToArray(), ct);
await twitterUsersBufferBlock.SendAsync(toSync, ct);
}
await Task.Delay(10, ct); // this is somehow necessary

14
src/BirdsiteLive.Pipeline/StatusPublicationPipeline.cs
View file

@ -18,7 +18,6 @@ namespace BirdsiteLive.Pipeline
public class StatusPublicationPipeline : IStatusPublicationPipeline
{
private readonly IRetrieveTwitterUsersProcessor _retrieveTwitterAccountsProcessor;
private readonly IRefreshTwitterUserStatusProcessor _refreshTwitterUserStatusProcessor;
private readonly IRetrieveTweetsProcessor _retrieveTweetsProcessor;
private readonly IRetrieveFollowersProcessor _retrieveFollowersProcessor;
private readonly ISendTweetsToFollowersProcessor _sendTweetsToFollowersProcessor;
@ -26,14 +25,13 @@ namespace BirdsiteLive.Pipeline
private readonly ILogger<StatusPublicationPipeline> _logger;
#region Ctor
public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, IRefreshTwitterUserStatusProcessor refreshTwitterUserStatusProcessor, ILogger<StatusPublicationPipeline> logger)
public StatusPublicationPipeline(IRetrieveTweetsProcessor retrieveTweetsProcessor, IRetrieveTwitterUsersProcessor retrieveTwitterAccountsProcessor, IRetrieveFollowersProcessor retrieveFollowersProcessor, ISendTweetsToFollowersProcessor sendTweetsToFollowersProcessor, ISaveProgressionProcessor saveProgressionProcessor, ILogger<StatusPublicationPipeline> logger)
{
_retrieveTweetsProcessor = retrieveTweetsProcessor;
_retrieveTwitterAccountsProcessor = retrieveTwitterAccountsProcessor;
_retrieveFollowersProcessor = retrieveFollowersProcessor;
_sendTweetsToFollowersProcessor = sendTweetsToFollowersProcessor;
_saveProgressionProcessor = saveProgressionProcessor;
_refreshTwitterUserStatusProcessor = refreshTwitterUserStatusProcessor;
_retrieveTwitterAccountsProcessor = retrieveTwitterAccountsProcessor;
_logger = logger;
}
@ -42,10 +40,8 @@ namespace BirdsiteLive.Pipeline
public async Task ExecuteAsync(CancellationToken ct)
{
// Create blocks
var twitterUserToRefreshBufferBlock = new BufferBlock<SyncTwitterUser[]>(new DataflowBlockOptions
var twitterUserToRefreshBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions
{ BoundedCapacity = 1, CancellationToken = ct });
var twitterUserToRefreshBlock = new TransformBlock<SyncTwitterUser[], UserWithDataToSync[]>(async x => await _refreshTwitterUserStatusProcessor.ProcessAsync(x, ct));
var twitterUsersBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 1, CancellationToken = ct });
var retrieveTweetsBlock = new TransformBlock<UserWithDataToSync[], UserWithDataToSync[]>(async x => await _retrieveTweetsProcessor.ProcessAsync(x, ct));
var retrieveTweetsBufferBlock = new BufferBlock<UserWithDataToSync[]>(new DataflowBlockOptions { BoundedCapacity = 2, CancellationToken = ct });
var retrieveFollowersBlock = new TransformManyBlock<UserWithDataToSync[], UserWithDataToSync>(async x => await _retrieveFollowersProcessor.ProcessAsync(x, ct));
@ -55,9 +51,7 @@ namespace BirdsiteLive.Pipeline
var saveProgressionBlock = new ActionBlock<UserWithDataToSync>(async x => await _saveProgressionProcessor.ProcessAsync(x, ct), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct });
// Link pipeline
twitterUserToRefreshBufferBlock.LinkTo(twitterUserToRefreshBlock, new DataflowLinkOptions { PropagateCompletion = true });
twitterUserToRefreshBlock.LinkTo(twitterUsersBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
twitterUsersBufferBlock.LinkTo(retrieveTweetsBlock, new DataflowLinkOptions { PropagateCompletion = true });
twitterUserToRefreshBufferBlock.LinkTo(retrieveTweetsBlock, new DataflowLinkOptions { PropagateCompletion = true });
retrieveTweetsBlock.LinkTo(retrieveTweetsBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });
retrieveTweetsBufferBlock.LinkTo(retrieveFollowersBlock, new DataflowLinkOptions { PropagateCompletion = true });
retrieveFollowersBlock.LinkTo(retrieveFollowersBufferBlock, new DataflowLinkOptions { PropagateCompletion = true });

128
src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RefreshTwitterUserStatusProcessorTests.cs
View file

@ -1,128 +0,0 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using BirdsiteLive.Common.Settings;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Moderation.Actions;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Processors;
using BirdsiteLive.Twitter;
using BirdsiteLive.Twitter.Models;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
namespace BirdsiteLive.Pipeline.Tests.Processors
{
[TestClass]
public class RefreshTwitterUserStatusProcessorTests
{
[TestMethod]
public async Task ProcessAsync_Test()
{
#region Stubs
var userId1 = 1;
var userId2 = 2;
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1
},
new SyncTwitterUser
{
Id = userId2
}
};
var settings = new InstanceSettings
{
FailingTwitterUserCleanUpThreshold = 300
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUserAsync(It.IsAny<string>()))
.ReturnsAsync(new TwitterUser
{
Protected = false
});
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(2 , result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
Assert.IsTrue(result.Any(x => x.User.Id == userId2));
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
[TestMethod]
public async Task ProcessAsync_ResetErrorCount_Test()
{
#region Stubs
var userId1 = 1;
var users = new List<SyncTwitterUser>
{
new SyncTwitterUser
{
Id = userId1,
FetchingErrorCount = 100
}
};
var settings = new InstanceSettings
{
FailingTwitterUserCleanUpThreshold = 300
};
#endregion
#region Mocks
var twitterUserServiceMock = new Mock<ICachedTwitterUserService>(MockBehavior.Strict);
twitterUserServiceMock
.Setup(x => x.GetUserAsync(It.IsAny<string>()))
.ReturnsAsync(new TwitterUser
{
Protected = false
});
var twitterUserDalMock = new Mock<ITwitterUserDal>(MockBehavior.Strict);
var removeTwitterAccountActionMock = new Mock<IRemoveTwitterAccountAction>(MockBehavior.Strict);
#endregion
var processor = new RefreshTwitterUserStatusProcessor(twitterUserServiceMock.Object, twitterUserDalMock.Object, removeTwitterAccountActionMock.Object, settings);
var result = await processor.ProcessAsync(users.ToArray(), CancellationToken.None);
#region Validations
Assert.AreEqual(1, result.Length);
Assert.IsTrue(result.Any(x => x.User.Id == userId1));
Assert.AreEqual(0, result.First().User.FetchingErrorCount);
twitterUserDalMock.VerifyAll();
removeTwitterAccountActionMock.VerifyAll();
#endregion
}
}
}

13
src/Tests/BirdsiteLive.Pipeline.Tests/Processors/RetrieveTwitterUsersProcessorTests.cs
View file

@ -8,6 +8,7 @@ using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Processors;
using BirdsiteLive.Pipeline.Tools;
using BirdsiteLive.Pipeline.Models;
using Microsoft.Extensions.Logging;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
@ -21,7 +22,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var users = new[]
{
new SyncTwitterUser(),
@ -64,7 +65,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_Multi_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var users = new List<SyncTwitterUser>();
for (var i = 0; i < 30; i++)
@ -110,7 +111,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_Multi2_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var users = new List<SyncTwitterUser>();
for (var i = 0; i < 31; i++)
@ -156,7 +157,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_NoUsers_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var maxUsers = 1000;
#endregion
@ -192,7 +193,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_Exception_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var maxUsers = 1000;
#endregion
@ -229,7 +230,7 @@ namespace BirdsiteLive.Pipeline.Tests.Processors
public async Task GetTwitterUsersAsync_Cancellation_Test()
{
#region Stubs
var buffer = new BufferBlock<SyncTwitterUser[]>();
var buffer = new BufferBlock<UserWithDataToSync[]>();
var canTokenS = new CancellationTokenSource();
canTokenS.Cancel();

14
src/Tests/BirdsiteLive.Pipeline.Tests/StatusPublicationPipelineTests.cs
View file

@ -1,7 +1,7 @@
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Contracts;
using Microsoft.Extensions.Logging;
using Microsoft.VisualStudio.TestTools.UnitTesting;
@ -20,14 +20,13 @@ namespace BirdsiteLive.Pipeline.Tests
#endregion
#region Mocks
var retrieveTwitterUsersProcessor = new Mock<IRetrieveTwitterUsersProcessor>(MockBehavior.Strict);
retrieveTwitterUsersProcessor
var retrieveTwitterUserProcessor = new Mock<IRetrieveTwitterUsersProcessor>(MockBehavior.Strict);
retrieveTwitterUserProcessor
.Setup(x => x.GetTwitterUsersAsync(
It.IsAny<BufferBlock<SyncTwitterUser[]>>(),
It.IsAny<BufferBlock<UserWithDataToSync[]>>(),
It.IsAny<CancellationToken>()))
.Returns(Task.Delay(0));
var refreshTwitterUserStatusProcessor = new Mock<IRefreshTwitterUserStatusProcessor>(MockBehavior.Strict);
var retrieveTweetsProcessor = new Mock<IRetrieveTweetsProcessor>(MockBehavior.Strict);
var retrieveFollowersProcessor = new Mock<IRetrieveFollowersProcessor>(MockBehavior.Strict);
var sendTweetsToFollowersProcessor = new Mock<ISendTweetsToFollowersProcessor>(MockBehavior.Strict);
@ -35,11 +34,10 @@ namespace BirdsiteLive.Pipeline.Tests
var logger = new Mock<ILogger<StatusPublicationPipeline>>();
#endregion
var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUsersProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, refreshTwitterUserStatusProcessor.Object, logger.Object);
var pipeline = new StatusPublicationPipeline(retrieveTweetsProcessor.Object, retrieveTwitterUserProcessor.Object, retrieveFollowersProcessor.Object, sendTweetsToFollowersProcessor.Object, saveProgressionProcessor.Object, logger.Object);
await pipeline.ExecuteAsync(ct.Token);
#region Validations
retrieveTwitterUsersProcessor.VerifyAll();
retrieveTweetsProcessor.VerifyAll();
retrieveFollowersProcessor.VerifyAll();
sendTweetsToFollowersProcessor.VerifyAll();