Care Pet ScyllaDB PHP IoT example¶

final class SimulateCommand extends AbstractCommand
{

    public function __construct(
        private readonly OwnerRepository $ownerRepository,
        private readonly PetRepository    $petRepository,
        private readonly SensorRepository $sensorRepository
    )
    {
    }

    const AMOUNT_BASE = 50000;

    public function __invoke(array $args = []): int
    {
        $this->info('Starting Sensor simulator...');
        foreach (range(0, self::AMOUNT_BASE) as $i) {
            $this->info("Batch: " . $i);
            [$ownerDTO, $petsDTO] = $this->generateFakeData();

            $this->ownerRepository->create($ownerDTO);
            $this->info(sprintf('Owner %s', $ownerDTO->id));

            $petsDTO->each(function ($petDTO) {
                $this->info(sprintf('Pet: %s | Owner %s', $petDTO->id->uuid(), $petDTO->ownerId));
                $this->petRepository->create($petDTO);

                SensorFactory::makeMany(5, ['pet_id' => $petDTO->id])
                    ->each($this->handleSensors());
            });
        }
        $this->info('Done :D');

        return self::SUCCESS;
    }

    private function generateFakeData(): array
    {
        $ownerDTO = OwnerFactory::make();
        $petsDTO = PetFactory::makeMany(5, ['owner_id' => $ownerDTO->id]);

        return [$ownerDTO, $petsDTO];
    }

    private function handleSensors(): Closure
    {
        return function (SensorDTO $sensorDTO) {
            $this->sensorRepository->create($sensorDTO);
            $this->info(sprintf(
                'Sensor: %s (%s) | Pet %s',
                $sensorDTO->id,
                $sensorDTO->type->name,
                $sensorDTO->petId
            ));
        };
    }
}

use Cassandra;
use Cassandra\Cluster;
use Cassandra\Cluster\Builder;
use Cassandra\FutureRows;
use Cassandra\Session;
use Cassandra\SimpleStatement;

class Connector
{
    public Builder $connectionBuilder;
    public Cluster $cluster;
    public Session $session;
    public SimpleStatement $query;

    const BASE_TIMEOUT = 10;

    public function __construct(array $config)
    {
        $this->connectionBuilder = Cassandra::cluster()
            ->withContactPoints($config['nodes'])
            ->withDefaultConsistency($config['consistency_level'])
            ->withPort($config['port']);

        if (!empty($config['username'] && !empty($config['password']))) {
            $this->connectionBuilder = $this->connectionBuilder->withCredentials($config['username'], $config['password']);
        }
        $this->cluster = $this->connectionBuilder->build();

        $this->session = $this->cluster->connect($config['keyspace']);
    }

    public function setKeyspace(string $keyspace = ''): self
    {
        $this->session->close(self::BASE_TIMEOUT);
        $this->session = $this->cluster->connect($keyspace);

        return $this;
    }

    public function prepare(string $query): self
    {
        $this->query = new SimpleStatement($query);

        return $this;
    }

    public function execute(): FutureRows
    {
        return $this->session->executeAsync($this->query, []);
    }
}

use App\Core\Entities\AbstractDTO;
use Cassandra\Rows;

abstract class AbstractRepository
{
    public string $table = '';

    public string $primaryKey = '';

    public Connector $connection;

    public array $keys = [];

    public function __construct(Connector $connector)
    {
        $this->connection = $connector;
    }

    public function getById(string $id): Rows
    {
        $query = sprintf("SELECT * FROM %s WHERE %s = %s", $this->table, $this->primaryKey, $id);

        return $this->connection
            ->prepare($query)
            ->execute()
            ->get(Connector::BASE_TIMEOUT);
    }

    public function all(): Rows
    {
        return $this->connection
            ->prepare(sprintf('SELECT * FROM %s', $this->table))
            ->execute()
            ->get(Connector::BASE_TIMEOUT);
    }

    public function create(AbstractDTO $dto): void
    {
        $keys = array_keys($dto->toDatabase());
        $dataValues = array_values($dto->toDatabase());

        foreach ($dataValues as $key => $value) {
            if (is_string($value) && !in_array($keys[$key], $this->keys)) {
                $value = addslashes($value);
                $dataValues[$key] = "'$value'";
            }
        }

        $query = sprintf(
            "INSERT INTO %s (%s) VALUES (%s)",
            $this->table,
            implode(', ', $keys),
            implode(', ', $dataValues)
        );


        $this->connection
            ->prepare($query)
            ->execute();
    }
}