我有一個S3存儲桶，mybucket并且想要在將新文件復制到該存儲桶中時執(zhí)行某些操作。對于通知，我想使用SQS隊列notifiqueue，因為我的目標是使用Laravel由于我在中創(chuàng)建了基礎(chǔ)架構(gòu)CloudFormation，因此資源的創(chuàng)建是這樣的：NotificationQueue: Type: AWS::SQS::Queue Properties: VisibilityTimeout: 120 QueueName: 'NotificationQueue'DataGateBucket: Type: AWS::S3::Bucket Properties: AccessControl: BucketOwnerFullControl BucketName: 'mybucket' NotificationConfiguration: QueueConfigurations: - Event: 's3:ObjectCreated:*' Queue: !GetAtt NotificationQueue.Arn每次將新文件保存在存儲桶中時，S3都會在SQS中自動創(chuàng)建一個通知?？杀氖牵行ж撦d的格式與Laravel標準作業(yè)有效負載不兼容，并且如果我在上運行輔助進程，則會NotificationQueue收到此錯誤：local.ERROR: Undefined index: job {"exception":"[object] (ErrorException(code: 0): Undefined index: job at .../vendor/laravel/framework/src/Illuminate/Queue/Jobs/Job.php:273)為了提供更完整的指示，這是我在通知中得到的內(nèi)容（將JSON轉(zhuǎn)換為PHP數(shù)組之后）array:1 [ "Records" => array:1 [ 0 => array:9 [ "eventVersion" => "2.1" "eventSource" => "aws:s3" "awsRegion" => "eu-central-1" "eventTime" => "2019-04-23T17:02:41.308Z" "eventName" => "ObjectCreated:Put" "userIdentity" => array:1 [ "principalId" => "AWS:XXXXXXXXXXXXXXXXXX" ] "requestParameters" => array:1 [ "sourceIPAddress" => "217.64.198.7" ] "responseElements" => array:2 [ "x-amz-request-id" => "602CE18B8DE0BE5C" "x-amz-id-2" => "wA/A3Jl2XpoxBWJEgQzy11s6O28Cz9Wc6pVi6Ho1vnIrOjqsWkGozlUmqRdpYAfub0MqdF8d/YI=" ] "s3" => array:4 [ "s3SchemaVersion" => "1.0" "configurationId" => "0d4eaa75-5730-495e-b6d4-368bf3690f30" "bucket" => array:3 [ "name" => "mybucket" "ownerIdentity" => array:1 [ "principalId" => "XXXXXXXXXXXXXXXXXX" ]使用Laravel訪問通知的最有效/正確/正確的方法是哪種，以便我可以觸發(fā)其他選項來響應(yīng)文件上傳？
1 回答
米琪卡哇伊
TA貢獻1998條經(jīng)驗 獲得超6個贊
我找到了一種獲取所需行為的方法，但是我不確定這是否是最佳方法，因此我將其發(fā)布在這里，也許可以給我反饋。
當我們談?wù)揕aravel Queues時，很多配置來自app.php，特別是來自本Provider節(jié)。我設(shè)法添加了我需要覆蓋原始QueueServiceProvider類的行為并替換了它：
// Here is the original Provider Class
//Illuminate\Queue\QueueServiceProvider::class,
// Here is the overridden Provider
\App\Providers\QueueServiceProvider::class,
新QueueServiceProvider類如下：
<?php
namespace App\Providers;
use App\Jobs\SqsNotifications\SqsConnector;
class QueueServiceProvider extends \Illuminate\Queue\QueueServiceProvider
{
/**
* Register the Amazon SQS queue connector.
*
* @param \Illuminate\Queue\QueueManager $manager
* @return void
*/
protected function registerSqsNotifConnector($manager)
{
$manager->addConnector('sqsNotif', function () {
return new SqsConnector();
});
}
public function registerConnectors($manager){
parent::registerConnectors($manager);
// Add the custom SQS notification connector
$this->registerSqsNotifConnector($manager);
}
}
請注意sqsNotif，需要將新的連接器添加到queue.php
'sqsNotif' => [
'driver' => 'sqsNotif',
'key' => env('AWS_ACCESS_KEY_ID'),
'secret' => env('AWS_SECRET_ACCESS_KEY'),
'prefix' => env('SQS_PREFIX', 'https://sqs.eu-central-1.amazonaws.com/your-account'),
'queue' => env('SQS_QUEUE', 'your-queue-name'),
'region' => env('AWS_DEFAULT_REGION', 'eu-central-1'),
],
在新版本中，QueueServiceProvider我們僅注冊了一個額外的連接器，其代碼為：
<?php
namespace App\Jobs\SqsNotifications;
use Aws\Sqs\SqsClient;
use Illuminate\Support\Arr;
class SqsConnector extends \Illuminate\Queue\Connectors\SqsConnector
{
/**
* Establish a queue connection.
*
* @param array $config
* @return \Illuminate\Contracts\Queue\Queue
*/
public function connect(array $config)
{
$config = $this->getDefaultConfiguration($config);
if ($config['key'] && $config['secret']) {
$config['credentials'] = Arr::only($config, ['key', 'secret', 'token']);
}
return new SqsQueue(
new SqsClient($config), $config['queue'], $config['prefix'] ?? ''
);
}
}
SqsQueue也以這種方式重新定義：
<?php
namespace App\Jobs\SqsNotifications;
class SqsQueue extends \Illuminate\Queue\SqsQueue
{
/**
* Pop the next job off of the queue.
*
* @param string $queue
* @return \Illuminate\Contracts\Queue\Job|null
*/
public function pop($queue = null)
{
$response = $this->sqs->receiveMessage([
'QueueUrl' => $queue = $this->getQueue($queue),
'AttributeNames' => ['ApproximateReceiveCount'],
]);
if (! is_null($response['Messages']) && count($response['Messages']) > 0) {
return new SqsJob(
$this->container, $this->sqs, $response['Messages'][0],
$this->connectionName, $queue
);
}
}
}
最后缺少的部分是SqsJob，其定義如下：
<?php
namespace App\Jobs\SqsNotifications;
use Illuminate\Queue\Jobs\JobName;
/**
* Class SqsJob
* @package App\Jobs\SqsNotifications
*
* Alternate SQS job that is used in case of S3 notifications
*/
class SqsJob extends \Illuminate\Queue\Jobs\SqsJob
{
/**
* Get the name of the queued job class.
*
* @return string
*/
public function getName()
{
$bucketName = '';
// Define the name of the Process based on the bucket name
switch($this->payload()['Records'][0]['s3']['bucket']['name']){
case 'mybucket':
$bucketName = 'NewMyBucketFileJob';
break;
}
return $bucketName;
}
/**
* Fire the job.
*
* @return void
*/
public function fire()
{
// Mimic the original behavior with a different payload
$payload = $this->payload();
[$class, $method] = JobName::parse('\App\Jobs\\' . $this->getName() . '@handle');
($this->instance = $this->resolve($class))->{$method}($payload);
// The Job wasn't automatically deleted, so we need to delete it manually once the process went fine
$this->delete();
}
}
在這一點上，我只需要在a中定義處理Job，例如下面的代碼NewMyBucketFileJob：
<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
class ProcessDataGateNewFile implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
/**
* Create a new job instance.
*
* @return void
*/
public function __construct()
{
}
/**
* Execute the job.
*
* @return void
*/
public function handle($data)
{
// Print the whole data structure
print_r($data);
// Or just the name of the uploaded file
print_r($data['Records'][0]['s3']['object']['key']);
}
}
這個過程是可行的，所以這是一個解決方案，但是涉及許多類擴展，并且在將來的版本中內(nèi)部隊列實現(xiàn)將被更改時，它非常脆弱。老實說，我想知道是否有更簡單或更強大的方法
- 1 回答
- 0 關(guān)注
- 166 瀏覽
添加回答
舉報
0/150
提交
取消