<commit_before><?php

namespace Rx\Operator;

use Rx\ObservableInterface;
use Rx\Observer\CallbackObserver;
use Rx\ObserverInterface;
use Rx\SchedulerInterface;

class SkipOperator implements OperatorInterface
{
    /**
     * @var integer
     */
    private $count;

    public function __construct($count)
    {
        if ($count < 0) {
            throw new \InvalidArgumentException('Count must be >= 0');
        }

        $this->count = $count;
    }

    /**
     * @param \Rx\ObservableInterface $observable
     * @param \Rx\ObserverInterface $observer
     * @param \Rx\SchedulerInterface $scheduler
     * @return \Rx\DisposableInterface
     */
    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
    {
        $remaining = $this->count;

        return $observable->subscribe(new CallbackObserver(
            function ($nextValue) use ($observer, &$remaining) {
                if ($remaining <= 0) {
                    $observer->onNext($nextValue);
                } else {
                    $remaining--;
                }
            },
            [$observer, 'onError'],
            [$observer, 'onCompleted']
        ));
    }
}<commit_msg>Pass scheduler through skip operator<commit_after><?php

namespace Rx\Operator;

use Rx\ObservableInterface;
use Rx\Observer\CallbackObserver;
use Rx\ObserverInterface;
use Rx\SchedulerInterface;

class SkipOperator implements OperatorInterface
{
    /**
     * @var integer
     */
    private $count;

    public function __construct($count)
    {
        if ($count < 0) {
            throw new \InvalidArgumentException('Count must be >= 0');
        }

        $this->count = $count;
    }

    /**
     * @param \Rx\ObservableInterface $observable
     * @param \Rx\ObserverInterface $observer
     * @param \Rx\SchedulerInterface $scheduler
     * @return \Rx\DisposableInterface
     */
    public function __invoke(ObservableInterface $observable, ObserverInterface $observer, SchedulerInterface $scheduler = null)
    {
        $remaining = $this->count;

        $cbObserver = new CallbackObserver(
            function ($nextValue) use ($observer, &$remaining) {
                if ($remaining <= 0) {
                    $observer->onNext($nextValue);
                } else {
                    $remaining--;
                }
            },
            [$observer, 'onError'],
            [$observer, 'onCompleted']
        );

        return $observable->subscribe($cbObserver, $scheduler);
    }
}
{'fraction_non_alphanumeric': 0.11567018366549434, 'fraction_numerical': 0.0023446658851113715, 'mean_word_length': 2.1219512195121952, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '500026', 'n_tokens_mistral': 722, 'n_tokens_neox': 661, 'n_words': 153}