glenux-archive/cloutier--bird.makeup
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

refactorization of sendtweets pipeline stage

Browse source
This commit is contained in:
Nicolas Constant 2020-08-12 19:05:01 -04:00
parent afa05a72d2
commit 4436b53319
No known key found for this signature in database
GPG key ID: 1E9F677FB01A5688
4 changed files with 156 additions and 93 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/BirdsiteLive.Pipeline/BirdsiteLive.Pipeline.csproj
View file

@ -1,4 +1,4 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework> <TargetFramework>netstandard2.0</TargetFramework>

101
src/BirdsiteLive.Pipeline/Processors/SendTweetsToFollowersProcessor.cs
View file

@ -10,6 +10,7 @@ using BirdsiteLive.DAL.Models;
using BirdsiteLive.Domain; using BirdsiteLive.Domain;
using BirdsiteLive.Pipeline.Contracts; using BirdsiteLive.Pipeline.Contracts;
using BirdsiteLive.Pipeline.Models; using BirdsiteLive.Pipeline.Models;
using BirdsiteLive.Pipeline.Processors.SubTasks;
using BirdsiteLive.Twitter; using BirdsiteLive.Twitter;
using BirdsiteLive.Twitter.Models; using BirdsiteLive.Twitter.Models;
using Tweetinvi.Models; using Tweetinvi.Models;
@ -18,16 +19,14 @@ namespace BirdsiteLive.Pipeline.Processors
{ {
public class SendTweetsToFollowersProcessor : ISendTweetsToFollowersProcessor public class SendTweetsToFollowersProcessor : ISendTweetsToFollowersProcessor
{ {
private readonly IActivityPubService _activityPubService; private readonly ISendTweetsToInboxTask _sendTweetsToInboxTask;
private readonly IStatusService _statusService; private readonly ISendTweetsToSharedInboxTask _sendTweetsToSharedInbox;
private readonly IFollowersDal _followersDal;
#region Ctor #region Ctor
public SendTweetsToFollowersProcessor(IActivityPubService activityPubService, IFollowersDal followersDal, IStatusService statusService) public SendTweetsToFollowersProcessor(ISendTweetsToInboxTask sendTweetsToInboxTask, ISendTweetsToSharedInboxTask sendTweetsToSharedInbox)
{ {
_activityPubService = activityPubService; _sendTweetsToInboxTask = sendTweetsToInboxTask;
_followersDal = followersDal; _sendTweetsToSharedInbox = sendTweetsToSharedInbox;
_statusService = statusService;
} }
#endregion #endregion
@ -58,7 +57,7 @@ namespace BirdsiteLive.Pipeline.Processors
{ {
try try
{ {
await ProcessInstanceFollowersWithSharedInbox(tweets, user, followersPerInstance); await _sendTweetsToSharedInbox.ExecuteAsync(tweets, user, followersPerInstance);
} }
catch (Exception e) catch (Exception e)
{ {
@ -67,58 +66,14 @@ namespace BirdsiteLive.Pipeline.Processors
} }
} }
} }
private async Task ProcessInstanceFollowersWithSharedInbox(ExtractedTweet[] tweets, SyncTwitterUser user,
IGrouping<string, Follower> followersPerInstance)
{
var userId = user.Id;
var host = followersPerInstance.Key;
var groupedFollowers = followersPerInstance.ToList();
var inbox = groupedFollowers.First().SharedInboxRoute;
var fromStatusId = groupedFollowers
.Max(x => x.FollowingsSyncStatus[userId]);
var tweetsToSend = tweets
.Where(x => x.Id > fromStatusId)
.OrderBy(x => x.Id)
.ToList();
var syncStatus = fromStatusId;
try
{
foreach (var tweet in tweetsToSend)
{
var note = _statusService.GetStatus(user.Acct, tweet);
var result =
await _activityPubService.PostNewNoteActivity(note, user.Acct, tweet.Id.ToString(), host, inbox);
if (result == HttpStatusCode.Accepted)
syncStatus = tweet.Id;
else
throw new Exception("Posting new note activity failed");
}
}
finally
{
if (syncStatus != fromStatusId)
{
foreach (var f in groupedFollowers)
{
f.FollowingsSyncStatus[userId] = syncStatus;
await _followersDal.UpdateFollowerAsync(f);
}
}
}
}
private async Task ProcessFollowersWithInbox(ExtractedTweet[] tweets, List<Follower> followerWtInbox, SyncTwitterUser user) private async Task ProcessFollowersWithInbox(ExtractedTweet[] tweets, List<Follower> followerWtInbox, SyncTwitterUser user)
{ {
foreach (var follower in followerWtInbox) foreach (var follower in followerWtInbox)
{ {
try try
{ {
await ProcessFollowerWithInboxAsync(tweets, follower, user); await _sendTweetsToInboxTask.ExecuteAsync(tweets, follower, user);
} }
catch (Exception e) catch (Exception e)
{ {
@ -127,43 +82,5 @@ namespace BirdsiteLive.Pipeline.Processors
} }
} }
} }
private async Task ProcessFollowerWithInboxAsync(IEnumerable<ExtractedTweet> tweets, Follower follower, SyncTwitterUser user)
{
var userId = user.Id;
var fromStatusId = follower.FollowingsSyncStatus[userId];
var tweetsToSend = tweets
.Where(x => x.Id > fromStatusId)
.OrderBy(x => x.Id)
.ToList();
var inbox = follower.InboxRoute;
//var inbox = string.IsNullOrWhiteSpace(follower.SharedInboxRoute)
// ? follower.InboxRoute
// : follower.SharedInboxRoute;
var syncStatus = fromStatusId;
try
{
foreach (var tweet in tweetsToSend)
{
var note = _statusService.GetStatus(user.Acct, tweet);
var result = await _activityPubService.PostNewNoteActivity(note, user.Acct, tweet.Id.ToString(), follower.Host, inbox);
if (result == HttpStatusCode.Accepted)
syncStatus = tweet.Id;
else
throw new Exception("Posting new note activity failed");
}
}
finally
{
if (syncStatus != fromStatusId)
{
follower.FollowingsSyncStatus[userId] = syncStatus;
await _followersDal.UpdateFollowerAsync(follower);
}
}
}
} }
} }

71
src/BirdsiteLive.Pipeline/Processors/SubTasks/SendTweetsToInboxTask.cs Normal file
View file

@ -0,0 +1,71 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Domain;
using BirdsiteLive.Twitter.Models;
namespace BirdsiteLive.Pipeline.Processors.SubTasks
{
public interface ISendTweetsToInboxTask
{
Task ExecuteAsync(IEnumerable<ExtractedTweet> tweets, Follower follower, SyncTwitterUser user);
}
public class SendTweetsToInboxTask : ISendTweetsToInboxTask
{
private readonly IActivityPubService _activityPubService;
private readonly IStatusService _statusService;
private readonly IFollowersDal _followersDal;
#region Ctor
public SendTweetsToInboxTask(IActivityPubService activityPubService, IStatusService statusService, IFollowersDal followersDal)
{
_activityPubService = activityPubService;
_statusService = statusService;
_followersDal = followersDal;
}
#endregion
public async Task ExecuteAsync(IEnumerable<ExtractedTweet> tweets, Follower follower, SyncTwitterUser user)
{
var userId = user.Id;
var fromStatusId = follower.FollowingsSyncStatus[userId];
var tweetsToSend = tweets
.Where(x => x.Id > fromStatusId)
.OrderBy(x => x.Id)
.ToList();
var inbox = follower.InboxRoute;
//var inbox = string.IsNullOrWhiteSpace(follower.SharedInboxRoute)
// ? follower.InboxRoute
// : follower.SharedInboxRoute;
var syncStatus = fromStatusId;
try
{
foreach (var tweet in tweetsToSend)
{
var note = _statusService.GetStatus(user.Acct, tweet);
var result = await _activityPubService.PostNewNoteActivity(note, user.Acct, tweet.Id.ToString(), follower.Host, inbox);
if (result == HttpStatusCode.Accepted)
syncStatus = tweet.Id;
else
throw new Exception("Posting new note activity failed");
}
}
finally
{
if (syncStatus != fromStatusId)
{
follower.FollowingsSyncStatus[userId] = syncStatus;
await _followersDal.UpdateFollowerAsync(follower);
}
}
}
}
}

75
src/BirdsiteLive.Pipeline/Processors/SubTasks/SendTweetsToSharedInboxTask.cs Normal file
View file

@ -0,0 +1,75 @@
using System;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using BirdsiteLive.DAL.Contracts;
using BirdsiteLive.DAL.Models;
using BirdsiteLive.Domain;
using BirdsiteLive.Twitter.Models;
namespace BirdsiteLive.Pipeline.Processors.SubTasks
{
public interface ISendTweetsToSharedInboxTask
{
Task ExecuteAsync(ExtractedTweet[] tweets, SyncTwitterUser user, IGrouping<string, Follower> followersPerInstance);
}
public class SendTweetsToSharedInboxTask : ISendTweetsToSharedInboxTask
{
private readonly IStatusService _statusService;
private readonly IActivityPubService _activityPubService;
private readonly IFollowersDal _followersDal;
#region Ctor
public SendTweetsToSharedInboxTask(IActivityPubService activityPubService, IStatusService statusService, IFollowersDal followersDal)
{
_activityPubService = activityPubService;
_statusService = statusService;
_followersDal = followersDal;
}
#endregion
public async Task ExecuteAsync(ExtractedTweet[] tweets, SyncTwitterUser user, IGrouping<string, Follower> followersPerInstance)
{
var userId = user.Id;
var host = followersPerInstance.Key;
var groupedFollowers = followersPerInstance.ToList();
var inbox = groupedFollowers.First().SharedInboxRoute;
var fromStatusId = groupedFollowers
.Max(x => x.FollowingsSyncStatus[userId]);
var tweetsToSend = tweets
.Where(x => x.Id > fromStatusId)
.OrderBy(x => x.Id)
.ToList();
var syncStatus = fromStatusId;
try
{
foreach (var tweet in tweetsToSend)
{
var note = _statusService.GetStatus(user.Acct, tweet);
var result =
await _activityPubService.PostNewNoteActivity(note, user.Acct, tweet.Id.ToString(), host, inbox);
if (result == HttpStatusCode.Accepted)
syncStatus = tweet.Id;
else
throw new Exception("Posting new note activity failed");
}
}
finally
{
if (syncStatus != fromStatusId)
{
foreach (var f in groupedFollowers)
{
f.FollowingsSyncStatus[userId] = syncStatus;
await _followersDal.UpdateFollowerAsync(f);
}
}
}
}
}
}