src/Centralstage/DeviceBundle/EventSubscriber/RawStoreMongoSubscriber.php line 20

Open in your IDE?
  1. <?php
  3. namespace App\Centralstage\DeviceBundle\EventSubscriber;
  5. use MongoDB\BSON\UTCDateTime;
  6. use App\Centralstage\CoreBundle\Utils\DateTimeHelper;
  7. use App\Centralstage\DeviceBundle\Models\DeviceDataModel;
  8. use App\Centralstage\DeviceBundle\Event\RawStoreMongoEvent;
  9. use Symfony\Component\DependencyInjection\ContainerInterface;
  10. use Symfony\Component\EventDispatcher\EventSubscriberInterface;
  12. class RawStoreMongoSubscriber implements EventSubscriberInterface
  13. {
  14.     private $primaryDocumentManager;
  15.     public function __construct(ContainerInterface $container)
  16.     {
  17.         $this->primaryDocumentManager   = $container->get('doctrine_mongodb.odm.document_manager');
  18.     }
  20.     public function onProcessDeviceRawData(RawStoreMongoEvent $RawObject) : void
  21.     {   
  22.         try {
  23.             $context = $RawObject->getData();
  24.             list($deviceChannelID, $data, $orgKey, $currentDate) = array_values($context);
  25.             $currentDT         = DateTimeHelper::parseDateTime($currentDate, 'Y-m-d', DateTimeHelper::ServerTimeZone);
  26.             $conDetails        = DeviceDataModel::getDeviceRawDBCollection($orgKey, $deviceChannelID, $currentDT);
  27.             $dataBase          = $conDetails["database"];
  28.             $collectionName    = $conDetails["collection"];
  29.             $data              = json_decode($data, true);
  30.             $deviceTimestamp   = $data['timestamp'];
  31.             $deviceTimestamp   = floor(strlen($deviceTimestamp) == 10 ? $deviceTimestamp*1000 : $deviceTimestamp);
  32.             $timestamp         = new UTCDateTime($deviceTimestamp); // Convert to milliseconds
  33.             $data['timeseries'] = $timestamp;
  34.             $data['timestamp'] = $deviceTimestamp;
  35.             $data['createdAt'] = new UTCDateTime();
  37.         
  38.             $database = $this->primaryDocumentManager->getClient()->selectDatabase($dataBase);
  39.         
  40.             // Check if collection exists and create if not
  41.             $collections = $database->listCollections(['filter' => ['name' => $collectionName]]);
  42.             if (iterator_count($collections) === 0) {
  43.                 $database->command([
  44.                     'create' => $collectionName,
  45.                     'timeseries' => [
  46.                         'timeField' => 'timeseries',
  47.                     ],
  48.                 ]);
  49.                 // Create index on the timestamp field
  50.                 $database->selectCollection($collectionName)->createIndex(['timestamp' => 1,'timeseries' => 1]);
  51.             }
  52.         
  53.             // Insert data into the collection
  54.             $database->selectCollection($collectionName)->insertOne($data);
  55.         } catch (\Exception $exception) {
  56.             $error = $exception->getMessage();
  57.         }
  58.     }
  59.     public static function getSubscribedEvents(): array
  60.     {
  61.         return [
  62.             'device.raw.data' => 'onProcessDeviceRawData'
  63.         ];
  64.     }
  66.     public function __destruct(){}
  67. }