PHP code example of mroosz / php-cassandra

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

mroosz / php-cassandra example snippets

use Cassandra\Connection;
use Cassandra\Connection\SocketNodeConfig;
use Cassandra\Connection\StreamNodeConfig;
use Cassandra\Consistency;

// Choose one or more nodes and a transport
$nodes = [
    new SocketNodeConfig(host: '127.0.0.1', port: 9042, username: 'cassandra', password: 'cassandra'),
    // or streams (supports SSL/TLS and persistent connections)
    // new StreamNodeConfig(host: '127.0.0.1', port: 9042, username: 'cassandra', password: 'cassandra'),
];

// Optional connection options (protocol STARTUP options)
$options = [
    // 'COMPRESSION' => 'lz4',
    // 'THROW_ON_OVERLOAD' => '1', // protocol v4+
];

$conn = new Connection($nodes, keyspace: 'my_keyspace', options: $options);
$conn->connect();

// Consistency default for subsequent requests
$conn->setConsistency(Consistency::QUORUM);

// Plain query (positional bind)
$rows = $conn->querySync(
    'SELECT * FROM users WHERE id = ?',
    [new \Cassandra\Type\Uuid('c5420d81-499e-4c9c-ac0c-fa6ba3ebc2bc')]
)->fetchAll();

// Prepared statement (named bind + paging)
$prepared = $conn->prepareSync('SELECT id, name FROM users WHERE org_id = :org_id');

$result = $conn->executeSync(
    $prepared,
    values: ['org_id' => 42],
    consistency: Consistency::LOCAL_QUORUM,
    options: new \Cassandra\Request\Options\ExecuteOptions(
        pageSize: 100,
        namesForValues: true
    )
);

foreach ($result as $row) {
    echo $row['name'], "\n";
}

use Cassandra\Connection\SocketNodeConfig;
use Cassandra\Connection\StreamNodeConfig;
use Cassandra\Connection;

$socketNode =new SocketNodeConfig(host: '10.0.0.10', port: 9042, username: 'user', password: 'secret',
        socketOptions: [SO_RCVTIMEO => ['sec' => 10, 'usec' => 0]]);

// Streams transport with SSL/TLS and persistent connection
$streamTlsNode = new StreamNodeConfig(
    host: 'tls://cassandra.example.com',
    port: 9042,
    username: 'user',
    password: 'secret',
    connectTimeoutInSeconds: 10,
    timeoutInSeconds: 30,
    persistent: true,
    sslOptions: [
        // See https://www.php.net/manual/en/context.ssl.php
        'cafile' => '/etc/ssl/certs/ca.pem',
        'verify_peer' => true,
        'verify_peer_name' => true,
    ]
);

$conn = new Connection([$socketNode, $tlsNode], keyspace: 'app');
$conn->connect();

$rowsResult = $conn->querySync(
    'SELECT id, name FROM users WHERE id = ?',
    [new \Cassandra\Type\Uuid($id)],
    consistency: \Cassandra\Consistency::ONE,
    options: new \Cassandra\Request\Options\QueryOptions(pageSize: 100)
);

$s1 = $conn->queryAsync('SELECT count(*) FROM t1');
$s2 = $conn->queryAsync('SELECT count(*) FROM t2');

$r2 = $s2->getResult();
$r1 = $s1->getResult();

$prepared = $conn->prepareSync('SELECT * FROM users WHERE email = :email');

$rowsResult = $conn->executeSync(
    $prepared,
    ['email' => '[email protected]'],
    options: new \Cassandra\Request\Options\ExecuteOptions(
        namesForValues: true,
        pageSize: 50
    )
);

$options = new \Cassandra\Request\Options\ExecuteOptions(pageSize: 100, namesForValues: true);
$result = $conn->executeSync($prepared, ['org_id' => 1], options: $options);

do {
    foreach ($result as $row) {
        // process row
    }

    $pagingState = $result->getMetadata()->pagingState;
    if ($pagingState === null) break;

    $options = new \Cassandra\Request\Options\ExecuteOptions(
        pageSize: 100,
        namesForValues: true,
        pagingState: $pagingState
    );
    $result = $conn->executeSync($result, [], options: $options); // reuse previous RowsResult for metadata id
} while (true);

use Cassandra\Request\Batch;
use Cassandra\Request\BatchType;

$batch = new Batch(type: BatchType::LOGGED, consistency: Consistency::QUORUM);

// Prepared in batch
$prepared = $conn->prepareSync('UPDATE users SET age = :age WHERE id = :id');
$batch->appendPreparedStatement($prepared, ['age' => 21, 'id' => 'c5419d81-499e-4c9c-ac0c-fa6ba3ebc2bc']);

// Simple query in batch
$batch->appendQuery(
    'INSERT INTO users (id, name, age) VALUES (?, ?, ?)',
    [
        new \Cassandra\Type\Uuid('c5420d81-499e-4c9c-ac0c-fa6ba3ebc2bc'),
        new \Cassandra\Type\Varchar('Mark'),
        20,
    ]
);

$conn->batchSync($batch);

use Cassandra\Response\Result\FetchType;

$r = $conn->querySync('SELECT role FROM system_auth.roles');
foreach ($r as $i => $row) {
    echo $row['role'], "\n";
}

$names = $r->fetchAllColumns(0); // remaining rows of first column

final class UserRow implements \Cassandra\Response\Result\RowClassInterface {
    public function __construct(private array $row, array $args = []) {}
    public function id(): string { return (string) $this->row['id']; }
    public function name(): string { return (string) $this->row['name']; }
}

$rows = $conn->querySync('SELECT id, name FROM users');
$rows->configureFetchObject(UserRow::class);

foreach ($rows as $user) {
    echo $user->name(), "\n";
}

// Scalars
new \Cassandra\Type\Ascii('hello');
new \Cassandra\Type\Bigint(10_000_000_000);
new \Cassandra\Type\Boolean(true);
new \Cassandra\Type\Double(2.718281828459);
new \Cassandra\Type\Float32(2.718);
new \Cassandra\Type\Integer(123);
new \Cassandra\Type\Smallint(2048);
new \Cassandra\Type\Tinyint(12);
new \Cassandra\Type\Varint(10000000000);

// Temporal
\Cassandra\Type\Date::fromString('2011-02-03');
\Cassandra\Type\Time::fromString('08:12:54.123456789');
\Cassandra\Type\Timestamp::fromString('2011-02-03T04:05:00.000+0000');
\Cassandra\Type\Duration::fromString('89h4m48s');

// Collections / Tuples / UDT
new \Cassandra\Type\CollectionList([1, 2, 3], [\Cassandra\Type::INT]);
new \Cassandra\Type\CollectionSet([1, 2, 3], [\Cassandra\Type::INT]);
new \Cassandra\Type\CollectionMap(['a' => 1], [\Cassandra\Type::ASCII, \Cassandra\Type::INT]);
new \Cassandra\Type\Tuple([1, 'x'], [\Cassandra\Type::INT, \Cassandra\Type::VARCHAR]);
new \Cassandra\Type\UDT(['id' => 1, 'name' => 'n'], ['id' => \Cassandra\Type::INT, 'name' => \Cassandra\Type::VARCHAR]);

new \Cassandra\Type\CollectionSet([
    [
        'id' => 1,
        'name' => 'string',
        'active' => true,
        'friends' => ['a', 'b'],
        'drinks' => [['qty' => 5, 'brand' => 'Pepsi']],
    ],
], [
    [
        'type' => \Cassandra\Type::UDT,
        'definition' => [
            'id' => \Cassandra\Type::INT,
            'name' => \Cassandra\Type::VARCHAR,
            'active' => \Cassandra\Type::BOOLEAN,
            'friends' => ['type' => \Cassandra\Type::COLLECTION_LIST, 'value' => \Cassandra\Type::VARCHAR],
            'drinks' => ['type' => \Cassandra\Type::COLLECTION_LIST, 'value' => [
                'type' => \Cassandra\Type::UDT,
                'typeMap' => ['qty' => \Cassandra\Type::INT, 'brand' => \Cassandra\Type::VARCHAR],
            ]],
        ],
    ],
]);

$conn->addEventListener(new class () implements \Cassandra\EventListener {
    public function onEvent(\Cassandra\Response\Event $event): void {
        // inspect $event->getType() and $event->getData()
    }
});

use Cassandra\Request\Register;
use Cassandra\Response\EventType;

$conn->syncRequest(new Register([
    EventType::TOPOLOGY_CHANGE->value,
    EventType::STATUS_CHANGE->value,
    EventType::SCHEMA_CHANGE->value,
]));

// process events (simplest possible loop)
while (true) {
    $conn->flush();
    sleep(1);
}

use Cassandra\Request\Query;

$req = new Query('SELECT now() FROM system.local');
$req->enableTracing();
$req->setPayload(['my-key' => 'my-value']);

$result = $conn->syncRequest($req);

$conn = new Cassandra\Connection($nodes, options: ['COMPRESSION' => 'lz4']);

bash
composer