public function query($query): Observable
    {
        return new AnonymousObservable(
            function (ObserverInterface $observer, SchedulerInterface $scheduler = null) use ($query) {
                if ($this->connStatus === $this::CONNECTION_NEEDED) {
                    $this->start();
                }
                if ($this->connStatus === $this::CONNECTION_BAD) {
                    $observer->onError(new \Exception('Connection failed'));
                    return new EmptyDisposable();
                }

                $q = new Query($query, $observer);
                $this->commandQueue[] = $q;
                $this->changeQueueCountInc();

                $this->processQueue();

                return new CallbackDisposable(function() use ($q) {
                    $this->changeQueueCountDec();
                    if ($this->currentCommand === $q && $q->isActive()) {
                        $this->cancelRequest();
                    }
                    $q->cancel();
                });
            }
        );

    }
public function query($query): Observable
    {
        return new AnonymousObservable(
            function (ObserverInterface $observer, SchedulerInterface $scheduler = null) use ($query) {
                if ($this->connStatus === $this::CONNECTION_NEEDED) {
                    $this->start();
                }
                if ($this->connStatus === $this::CONNECTION_BAD) {
                    $observer->onError(new \Exception('Connection failed'));
                    return new EmptyDisposable();
                }

                $q = new Query($query, $observer);
                $this->commandQueue[] = $q;
                $this->changeQueueCountInc();

                $this->processQueue();

                return new CallbackDisposable(function() use ($q) {
                    $this->changeQueueCountDec();
                    if ($this->currentCommand === $q && $q->isActive()) {
                        $this->cancelRequest();
                    }
                    $q->cancel();
                });
            }
        );

    }
public function query ( $ query ) : Observable { return new AnonymousObservable ( function ( ObserverInterface $ observer , SchedulerInterface $ scheduler = null ) use ( $ query ) { if ( $ this -> connStatus === $ this :: CONNECTION_NEEDED ) { $ this -> start ( ) ; } if ( $ this -> connStatus === $ this :: CONNECTION_BAD ) { $ observer -> onError ( new \ Exception ( 'Connection failed' ) ) ; return new EmptyDisposable ( ) ; } $ q = new Query ( $ query , $ observer ) ; $ this -> commandQueue [ ] = $ q ; $ this -> changeQueueCountInc ( ) ; $ this -> processQueue ( ) ; return new CallbackDisposable ( function ( ) use ( $ q ) { $ this -> changeQueueCountDec ( ) ; if ( $ this -> currentCommand === $ q && $ q -> isActive ( ) ) { $ this -> cancelRequest ( ) ; } $ q -> cancel ( ) ; } ) ; } ) ; }
Execute simple query
@param string $query
@return Observable
Execute simple query