WRStatsWriter.php

Go to the documentation of this file.

<?php
 
namespace Wikimedia\WRStats;
 
class WRStatsWriter {
    private $store;
    private $metricSpecs;
    private $queuedValues = [];
    private $now;
    private $prefixComponents;
 
    public function __construct( StatsStore $store, $specs, $prefix ) {
        $this->store = $store;
        $this->metricSpecs = [];
        foreach ( $specs as $name => $spec ) {
            $this->metricSpecs[$name] = new MetricSpec( $spec );
        }
        $this->prefixComponents = is_array( $prefix ) ? $prefix : [ $prefix ];
        if ( !count( $this->prefixComponents ) ) {
            throw new WRStatsError( __METHOD__ .
                ': there must be at least one prefix component' );
        }
    }
 
    public function incr( $name, ?EntityKey $entity = null, $value = 1 ) {
        $metricSpec = $this->metricSpecs[$name] ?? null;
        $entity = $entity ?? new LocalEntityKey;
        if ( $metricSpec === null ) {
            throw new WRStatsError( __METHOD__ . ": Unrecognised metric \"$name\"" );
        }
        $res = $metricSpec->resolution;
        $scaledValue = $value / $res;
 
        foreach ( $metricSpec->sequences as $seqSpec ) {
            $timeStep = $seqSpec->timeStep;
            $timeBucket = (int)( $this->now() / $timeStep );
            $key = $this->store->makeKey(
                $this->prefixComponents,
                [ $name, $seqSpec->name, $timeBucket ],
                $entity
            );
 
            $ttl = $seqSpec->hardExpiry;
 
            if ( !isset( $this->queuedValues[$ttl][$key] ) ) {
                $this->queuedValues[$ttl][$key] = 0;
            }
            $this->queuedValues[$ttl][$key] += (int)round( $scaledValue );
        }
    }
 
    public function setCurrentTime( $now ) {
        $this->now = $now;
    }
 
    public function resetCurrentTime() {
        $this->now = null;
    }
 
    private function now() {
        if ( $this->now === null ) {
            $this->now = microtime( true );
        }
        return $this->now;
    }
 
    public function flush() {
        foreach ( $this->queuedValues as $ttl => $values ) {
            $this->store->incr( $values, $ttl );
        }
        $this->queuedValues = [];
    }
 
    public function __destruct() {
        $this->flush();
    }
 
    public function resetAll( ?array $entities = null ) {
        $entities = $entities ?? [ new LocalEntityKey ];
        $this->queuedValues = [];
        $keys = [];
        foreach ( $this->metricSpecs as $name => $metricSpec ) {
            foreach ( $metricSpec->sequences as $seqSpec ) {
                $timeStep = $seqSpec->timeStep;
                $ttl = $seqSpec->hardExpiry;
                $lastBucket = (int)( $this->now() / $timeStep ) + 1;
                $firstBucket = (int)( ( $this->now() - $ttl ) / $timeStep ) - 1;
                for ( $bucket = $firstBucket; $bucket <= $lastBucket; $bucket++ ) {
                    foreach ( $entities as $entity ) {
                        $keys[] = $this->store->makeKey(
                            $this->prefixComponents,
                            [ $name, $seqSpec->name, $bucket ],
                            $entity
                        );
                    }
                }
            }
        }
        $this->store->delete( $keys );
    }
}