function BatchSpanProcessor::flush

3 calls to BatchSpanProcessor::flush()

BatchSpanProcessor::forceFlush in vendor/open-telemetry/sdk/Trace/SpanProcessor/BatchSpanProcessor.php

Export all ended spans to the configured Exporter that have not yet been exported. Returns `true` if the flush was successful, otherwise `false`.

BatchSpanProcessor::onEnd in vendor/open-telemetry/sdk/Trace/SpanProcessor/BatchSpanProcessor.php

BatchSpanProcessor::shutdown in vendor/open-telemetry/sdk/Trace/SpanProcessor/BatchSpanProcessor.php

Cleanup; after shutdown, calling onStart, onEnd, or forceFlush is invalid Returns `false` is the processor is already shutdown, otherwise `true`.

File

vendor/open-telemetry/sdk/Trace/SpanProcessor/BatchSpanProcessor.php, line 198

Class

BatchSpanProcessor

Namespace

OpenTelemetry\SDK\Trace\SpanProcessor

Code

private function flush(?string $flushMethod = null, ?CancellationInterface $cancellation = null) : bool {
    if ($flushMethod !== null) {
        $flushId = $this->batchId + $this->queue
            ->count() + (int) (bool) $this->batch;
        $this->flush
            ->enqueue([
            $flushId,
            $flushMethod,
            $cancellation,
            !$this->running,
            Context::getCurrent(),
        ]);
    }
    if ($this->running) {
        return false;
    }
    $success = true;
    $exception = null;
    $this->running = true;
    try {
        for (;;) {
            while (!$this->flush
                ->isEmpty() && $this->flush
                ->bottom()[0] <= $this->batchId) {
                [
                    ,
                    $flushMethod,
                    $cancellation,
                    $propagateResult,
                    $context,
                ] = $this->flush
                    ->dequeue();
                $scope = $context->activate();
                try {
                    $result = $this->exporter
                        ->{$flushMethod}($cancellation);
                    if ($propagateResult) {
                        $success = $result;
                    }
                } catch (Throwable $e) {
                    if ($propagateResult) {
                        $exception = $e;
                    }
                    else {
                        self::logError(sprintf('Unhandled %s error', $flushMethod), [
                            'exception' => $e,
                        ]);
                    }
                } finally {
                    $scope->detach();
                }
            }
            if (!$this->shouldFlush()) {
                break;
            }
            if ($this->queue
                ->isEmpty()) {
                $this->enqueueBatch();
            }
            $batchSize = count($this->queue
                ->bottom());
            $this->batchId++;
            $scope = $this->exportContext
                ->activate();
            try {
                $this->exporter
                    ->export($this->queue
                    ->dequeue())
                    ->await();
            } catch (Throwable $e) {
                self::logError('Unhandled export error', [
                    'exception' => $e,
                ]);
            } finally {
                $this->processed += $batchSize;
                $this->queueSize -= $batchSize;
                $scope->detach();
            }
        }
    } finally {
        $this->running = false;
    }
    if ($exception !== null) {
        throw $exception;
    }
    return $success;
}