AvroFormatter.php

Go to the documentation of this file.

<?php
namespace MediaWiki\Logger\Monolog;
 
use AvroIODatumWriter;
use AvroIOBinaryEncoder;
use AvroIOTypeException;
use AvroSchema;
use AvroStringIO;
use AvroValidator;
use Monolog\Formatter\FormatterInterface;
 
class AvroFormatter implements FormatterInterface {
    const MAGIC = 0x0;
    protected $schemas;
 
    protected $io;
 
    protected $encoder;
 
    protected $writer;
 
    public function __construct( array $schemas ) {
        $this->schemas = $schemas;
        $this->io = new AvroStringIO( '' );
        $this->encoder = new AvroIOBinaryEncoder( $this->io );
        $this->writer = new AvroIODatumWriter();
    }
 
    public function format( array $record ) {
        $this->io->truncate();
        $schema = $this->getSchema( $record['channel'] );
        $revId = $this->getSchemaRevisionId( $record['channel'] );
        if ( $schema === null || $revId === null ) {
            trigger_error( "The schema for channel '{$record['channel']}' is not available" );
            return null;
        }
        try {
            $this->writer->write_data( $schema, $record['context'], $this->encoder );
        } catch ( AvroIOTypeException $e ) {
            $errors = AvroValidator::getErrors( $schema, $record['context'] );
            $json = json_encode( $errors );
            trigger_error( "Avro failed to serialize record for {$record['channel']} : {$json}" );
            return null;
        }
        return chr( self::MAGIC ) . $this->encodeLong( $revId ) . $this->io->string();
    }
 
    public function formatBatch( array $records ) {
        $result = [];
        foreach ( $records as $record ) {
            $message = $this->format( $record );
            if ( $message !== null ) {
                $result[] = $message;
            }
        }
        return $result;
    }
 
    protected function getSchema( $channel ) {
        if ( !isset( $this->schemas[$channel] ) ) {
            return null;
        }
        if ( !isset( $this->schemas[$channel]['revision'], $this->schemas[$channel]['schema'] ) ) {
            return null;
        }
 
        if ( !$this->schemas[$channel]['schema'] instanceof AvroSchema ) {
            $schema = $this->schemas[$channel]['schema'];
            if ( is_string( $schema ) ) {
                $this->schemas[$channel]['schema'] = AvroSchema::parse( $schema );
            } else {
                $this->schemas[$channel]['schema'] = AvroSchema::real_parse(
                    $schema
                );
            }
        }
        return $this->schemas[$channel]['schema'];
    }
 
    public function getSchemaRevisionId( $channel ) {
        if ( isset( $this->schemas[$channel]['revision'] ) ) {
            return (int)$this->schemas[$channel]['revision'];
        }
        return null;
    }
 
    private function encodeLong( $id ) {
        $high   = ( $id & 0xffffffff00000000 ) >> 32;
        $low    = $id & 0x00000000ffffffff;
        return pack( 'NN', $high, $low );
    }
}