mirror of
https://github.com/pixelfed/pixelfed.git
synced 2024-11-22 14:31:26 +00:00
Add Feed fanout
This commit is contained in:
parent
de2b5ba4e9
commit
ce63c4997b
4 changed files with 107 additions and 18 deletions
|
@ -10,8 +10,6 @@ use Illuminate\Queue\InteractsWithQueue;
|
|||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Queue\Middleware\WithoutOverlapping;
|
||||
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
|
||||
use App\Status;
|
||||
use App\Follower;
|
||||
use App\Services\FollowerService;
|
||||
use App\Services\HomeTimelineService;
|
||||
|
||||
|
@ -19,7 +17,8 @@ class FeedInsertPipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
|
|||
{
|
||||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||||
|
||||
protected $status;
|
||||
protected $sid;
|
||||
protected $pid;
|
||||
|
||||
public $timeout = 900;
|
||||
public $tries = 3;
|
||||
|
@ -38,7 +37,7 @@ class FeedInsertPipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
|
|||
*/
|
||||
public function uniqueId(): string
|
||||
{
|
||||
return 'hfp:f-insert:sid:' . $this->status->id;
|
||||
return 'hts:feed:insert:sid:' . $this->sid;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -48,15 +47,16 @@ class FeedInsertPipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
|
|||
*/
|
||||
public function middleware(): array
|
||||
{
|
||||
return [(new WithoutOverlapping("hfp:f-insert:sid:{$this->status->id}"))->shared()->dontRelease()];
|
||||
return [(new WithoutOverlapping("hts:feed:insert:sid:{$this->sid}"))->shared()->dontRelease()];
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new job instance.
|
||||
*/
|
||||
public function __construct(Status $status)
|
||||
public function __construct($sid, $pid)
|
||||
{
|
||||
$this->status = $status;
|
||||
$this->sid = $sid;
|
||||
$this->pid = $pid;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -64,13 +64,10 @@ class FeedInsertPipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
|
|||
*/
|
||||
public function handle(): void
|
||||
{
|
||||
$status = $this->status;
|
||||
$sid = $status->id;
|
||||
$pid = $status->profile_id;
|
||||
$ids = FollowerService::localFollowerIds($pid);
|
||||
$ids = FollowerService::localFollowerIds($this->pid);
|
||||
|
||||
foreach($ids as $id) {
|
||||
HomeTimelineService::add($id, $sid);
|
||||
HomeTimelineService::add($id, $this->sid);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
73
app/Jobs/HomeFeedPipeline/FeedRemovePipeline.php
Normal file
73
app/Jobs/HomeFeedPipeline/FeedRemovePipeline.php
Normal file
|
@ -0,0 +1,73 @@
|
|||
<?php
|
||||
|
||||
namespace App\Jobs\HomeFeedPipeline;
|
||||
|
||||
use Illuminate\Bus\Queueable;
|
||||
use Illuminate\Contracts\Queue\ShouldBeUnique;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
use Illuminate\Foundation\Bus\Dispatchable;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Queue\Middleware\WithoutOverlapping;
|
||||
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
|
||||
use App\Services\FollowerService;
|
||||
use App\Services\HomeTimelineService;
|
||||
|
||||
class FeedRemovePipeline implements ShouldQueue, ShouldBeUniqueUntilProcessing
|
||||
{
|
||||
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
|
||||
|
||||
protected $sid;
|
||||
protected $pid;
|
||||
|
||||
public $timeout = 900;
|
||||
public $tries = 3;
|
||||
public $maxExceptions = 1;
|
||||
public $failOnTimeout = true;
|
||||
|
||||
/**
|
||||
* The number of seconds after which the job's unique lock will be released.
|
||||
*
|
||||
* @var int
|
||||
*/
|
||||
public $uniqueFor = 3600;
|
||||
|
||||
/**
|
||||
* Get the unique ID for the job.
|
||||
*/
|
||||
public function uniqueId(): string
|
||||
{
|
||||
return 'hts:feed:remove:sid:' . $this->sid;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the middleware the job should pass through.
|
||||
*
|
||||
* @return array<int, object>
|
||||
*/
|
||||
public function middleware(): array
|
||||
{
|
||||
return [(new WithoutOverlapping("hts:feed:remove:sid:{$this->sid}"))->shared()->dontRelease()];
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new job instance.
|
||||
*/
|
||||
public function __construct($sid, $pid)
|
||||
{
|
||||
$this->sid = $sid;
|
||||
$this->pid = $pid;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the job.
|
||||
*/
|
||||
public function handle(): void
|
||||
{
|
||||
$ids = FollowerService::localFollowerIds($this->pid);
|
||||
|
||||
foreach($ids as $id) {
|
||||
HomeTimelineService::rem($id, $this->sid);
|
||||
}
|
||||
}
|
||||
}
|
|
@ -21,6 +21,8 @@ use Illuminate\Queue\InteractsWithQueue;
|
|||
use Illuminate\Queue\SerializesModels;
|
||||
use App\Services\UserFilterService;
|
||||
use App\Services\AdminShadowFilterService;
|
||||
use App\Jobs\HomeFeedPipeline\FeedInsertPipeline;
|
||||
use App\Jobs\HomeFeedPipeline\HashtagInsertFanoutPipeline;
|
||||
|
||||
class StatusEntityLexer implements ShouldQueue
|
||||
{
|
||||
|
@ -105,12 +107,12 @@ class StatusEntityLexer implements ShouldQueue
|
|||
}
|
||||
DB::transaction(function () use ($status, $tag) {
|
||||
$slug = str_slug($tag, '-', false);
|
||||
$hashtag = Hashtag::where('slug', $slug)->first();
|
||||
if (!$hashtag) {
|
||||
$hashtag = Hashtag::create(
|
||||
['name' => $tag, 'slug' => $slug]
|
||||
);
|
||||
}
|
||||
|
||||
$hashtag = Hashtag::firstOrCreate([
|
||||
'slug' => $slug
|
||||
], [
|
||||
'name' => $tag
|
||||
]);
|
||||
|
||||
StatusHashtag::firstOrCreate(
|
||||
[
|
||||
|
@ -150,6 +152,18 @@ class StatusEntityLexer implements ShouldQueue
|
|||
MentionPipeline::dispatch($status, $m);
|
||||
});
|
||||
}
|
||||
$this->fanout();
|
||||
}
|
||||
|
||||
public function fanout()
|
||||
{
|
||||
$status = $this->status;
|
||||
|
||||
if(config('exp.cached_home_timeline')) {
|
||||
if($status->in_reply_to_id == null && in_array($status->scope, ['public', 'unlisted', 'private'])) {
|
||||
FeedInsertPipeline::dispatch($status->id, $status->profile_id)->onQueue('feed');
|
||||
}
|
||||
}
|
||||
$this->deliver();
|
||||
}
|
||||
|
||||
|
|
|
@ -7,6 +7,7 @@ use App\Services\ProfileStatusService;
|
|||
use Cache;
|
||||
use App\Models\ImportPost;
|
||||
use App\Services\ImportService;
|
||||
use App\Jobs\HomeFeedPipeline\FeedRemovePipeline;
|
||||
|
||||
class StatusObserver
|
||||
{
|
||||
|
@ -63,6 +64,10 @@ class StatusObserver
|
|||
ImportPost::whereProfileId($status->profile_id)->whereStatusId($status->id)->delete();
|
||||
ImportService::clearImportedFiles($status->profile_id);
|
||||
}
|
||||
|
||||
if(config('exp.cached_home_timeline')) {
|
||||
FeedRemovePipeline::dispatch($status->id, $status->profile_id)->onQueue('feed');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
Loading…
Reference in a new issue