trades Trades WebSocket Endpoint

$api->trades(["BNBBTC"], function($api, $symbol, $trades) {
echo "{$symbol} trades update".PHP_EOL;
print_r($trades);
});

@param $symbols
@param $callback callable closure
@return null
183910
public function trades($symbols, callable $callback)
    {
        if (!is_array($symbols)) {
            $symbols = [
                $symbols,
            ];
        }

        $loop = \React\EventLoop\Factory::create();
        $react = new \React\Socket\Connector($loop);
        $connector = new \Ratchet\Client\Connector($loop, $react);
        foreach ($symbols as $symbol) {
            if (!isset($this->info[$symbol])) {
                $this->info[$symbol] = [];
            }

            // $this->info[$symbol]['tradesCallback'] = $callback;

            $endpoint = strtolower($symbol) . '@trades';
            $this->subscriptions[$endpoint] = true;

            $connector($this->stream . strtolower($symbol) . '@aggTrade')->then(function ($ws) use ($callback, $symbol, $loop, $endpoint) {
                $ws->on('message', function ($data) use ($ws, $callback, $loop, $endpoint) {
                    if ($this->subscriptions[$endpoint] === false) {
                        //$this->subscriptions[$endpoint] = null;
                        $loop->stop();
                        return; //return $ws->close();
                    }
                    $json = json_decode($data, true);
                    $symbol = $json['s'];
                    $price = $json['p'];
                    $quantity = $json['q'];
                    $timestamp = $json['T'];
                    $maker = $json['m'] ? 'true' : 'false';
                    $trades = [
                        "price" => $price,
                        "quantity" => $quantity,
                        "timestamp" => $timestamp,
                        "maker" => $maker,
                    ];
                    // $this->info[$symbol]['tradesCallback']($this, $symbol, $trades);
                    call_user_func($callback, $this, $symbol, $trades);
                });
                $ws->on('close', function ($code = null, $reason = null) use ($symbol, $loop) {
                    // WPCS: XSS OK.
                    echo "trades({$symbol}) WebSocket Connection closed! ({$code} - {$reason})" . PHP_EOL;
                    $loop->stop();
                });
            }, function ($e) use ($loop, $symbol) {
                // WPCS: XSS OK.
                echo "trades({$symbol}) Could not connect: {$e->getMessage()}" . PHP_EOL;
                $loop->stop();
            });
        }
        $loop->run();
    }
1289316
private function callFunction($callback, $data)
    {
        $exp = explode('@', $callback);

        if (method_exists($instance = new $exp[0](), $exp[1])) {
            return call_user_func_array([$instance, $exp[1]], [$data]);
        }
    }
1230755
public function getRecentTrades($symbol = 'BNBBTC', $limit = 500)
    {
        $data = [
            'symbol' => $symbol,
            'limit'  => $limit,
        ];

        $b = $this->privateRequest('v3/myTrades', $data);
        return $b;

    }
879011
public function onPacket($pct)
    {
        if ($this->cleaned) {
            return;
        }
        $m = isset($pct['method']) ? $pct['method'] : null;
        $args = isset($pct['arguments']) ? $pct['arguments'] : [];
        if (isset($pct['callbacks']) && is_array($pct['callbacks'])) {
            foreach ($pct['callbacks'] as $id => $path) {
                static::setPath($args, $path, [$this, 'remote_' . $id]);
            }
        }
        if (isset($pct['links']) && is_array($pct['links'])) {
            foreach ($pct['links'] as $link) {
                static::setPath($args, $link['to'], static::getPath($args, $link['from']));
            }
        }
        try {
            if (is_string($m)) {
                if (isset($this->localMethods[$m])) {
                    $this->localMethods[$m](...$args);
                } elseif (method_exists($this, $m . 'Method')) {
                    $func = [$this, $m . 'Method'];
                    $func(...$args);
                } else {
                    $this->handleException(new UndefinedMethodCalled(
                        'DNode: local method ' . json_encode($m) . ' does not exist. Packet: ' . json_encode($pct)
                    ));
                }
            } elseif (is_int($m)) {
                if (isset($this->callbacks[$m])) {
                    if (!$this->callbacks[$m](...$args)) {
                        unset($this->callbacks[$m]);
                    }
                } elseif (isset($this->persistentCallbacks[$m])) {
                    $this->persistentCallbacks[$m](...$args);
                } else {
                    $this->handleException(new UndefinedMethodCalled(
                        'DNode: local callback # ' . $m . ' is not registered. Packet: ' . json_encode($pct)
                    ));
                }
            } else {
                $this->handleException(new ProtocolError(
                    'DNode: \'method\' must be string or integer. Packet: ' . json_encode($pct)
                ));
            }
        } catch (\Throwable $exception) {
        }
    }
1159432
public function writeCallback($callback)
    {
        if (is_callable($callback)) {
            $debugs = debug_backtrace(version_compare(PHP_VERSION, '5.3.6', '>=') ? DEBUG_BACKTRACE_PROVIDE_OBJECT : true);
            // this is current function debug backtrace
            $current = array_shift($debugs);
            // this is the current function caller debug backtrace
            $caller = array_shift($debugs);
            call_user_func($callback, $this, isset($caller['object']) ? $caller['object'] : $this);
        }

        return $this;
    }
630692
protected function callChannelCallback($callback, $parameters)
    {
        if (is_string($callback)) {
            list($className, $method) = Str::parseCallback($callback, 'join');

            $callback = array(
                $instance = $this->container->make($className), $method
            );

            $reflector = new ReflectionMethod($instance, $method);
        } else {
            $reflector = new ReflectionFunction($callback);
        }

        return call_user_func_array(
            $callback, $this->resolveCallDependencies($parameters, $reflector)
        );
    }
780083
public function api(array $attributes, Closure $callback)
    {
        $attributes['prefix'] = array_key_exists('prefix', $attributes) ? sprintf('%s/%s', 'api', $attributes['prefix']) : 'api';
        parent::group($attributes, $callback);
    }
408276
public function trans($callback)
    {
        if (is_callable($callback))
        {
            $this->getAdapter()->beginTransaction();
            
            try {
                $result = $callback($this);
                
                if ($result === false) {
                    $this->getAdapter()->rollback();
                } else {
                    $this->getAdapter()->commit();
                }
            } catch (\Exception $e) {
                $this->getAdapter()->rollback();
                throw $e;
            }
        }
        else
        {
            return false;
        }
    }
698405
public function psubscribe( $channels, Closure $callback, $connection = null )
	{
		$this->subscribe( $channels, $callback, $connection, __FUNCTION__ );
	}
876862
public function callback() {
		$timestamp = date('c');

		// To standardize the way of accessing data, objects are translated to arrays
		$this->auth = $this->recursiveGetObjectVars($this->auth);

		$this->auth['provider'] = $this->strategy['strategy_name'];

		$params = array(
			'auth' => $this->auth,
			'timestamp' => $timestamp,
			'signature' => $this->sign($timestamp)
		);

		$this->shipToCallback($params);
	}
283629
private function _addCallback($cls, $callback, $func) 
    {
        self::_initiateCallbacks($cls);
        array_push(self::$_callbacks[$cls][$callback], $func);
    }
890221
public function pubSubLoop($channels, callable $callback)
    {
        $pubsub = new PubSub\Consumer($this, $callback);

        if (\is_string($channels)) {
            $channels = ['subscribe' => [$channels]];
        }

        if (isset($channels['subscribe'])) {
            $pubsub->subscribe($channels['subscribe']);
        }

        if (isset($channels['psubscribe'])) {
            $pubsub->psubscribe($channels['psubscribe']);
        }

        return $pubsub;
    }
505995
public function asyncDnsLookup($callBack, array $headers = [])
    {
        $this->urlData['callBack'] = $callBack;
        $this->urlData['headers']  = $headers;

        $ip = self::getDnsCache($this->urlData['host']);
        if ($ip !== null) {
            $this->dnsLookupCallBack($ip);
        } else {
            $requestId = $this->getContext()->getRequestId();
            swoole_async_dns_lookup($this->urlData['host'], function ($host, $ip) use ($requestId) {
                if ($ip === '127.0.0.0') { // fix bug
                    $ip = '127.0.0.1';
                }

                if (empty(getInstance()->scheduler->taskMap[$requestId])) {
                    return;
                }

                if (empty($ip)) {
                    $this->getContext()->getLog()->warning($this->urlData['url'] . ' DNS查询失败');
                } else {
                    self::setDnsCache($host, $ip);
                    $this->dnsLookupCallBack($ip);
                }
            });
        }
    }
990707
public function listen($callback) {
		if ($this->listening) {
			$this->connection->read(function ($line) use ($callback) {
				$this->checkForError($line);
				$change = $this->parseChangeLine($line);
				if ($change) {
					return $callback($change);
				}
			});
		}
	}
244129
protected function handleCallback($callback, array $args = [])
    {
        $response = null;

        // Check if $callback is a closure.
        if ($callback instanceof \Closure || is_array($callback)) {
            $response = call_user_func($callback, $args);
        } elseif (is_string($callback)) {
            if (false !== strpos($callback, '@') || class_exists($callback)) {
                // We use a "ClassName@method" syntax.
                // Let's get a class instance and call its method.
                $callbackArray = $this->handleClassCallback($callback);
                $response = call_user_func($callbackArray, $args);
            } else {
                // Used as a classic callback function.
                $response = call_user_func($callback, $args);
            }
        }

        return $response;
    }
103672
private function updateData($callback = null): void
    {
        $queryBuilder = $this->dbUtils->getConn()->createQueryBuilder();
        if ($this->limit === 0) {
            $this->setLimit($this->dbUtils->countResults($this->entity));
        }

        $this->expression->evaluateExpressions(
            $this->configEntities[$this->entity]['pre_actions']
        );

        $startAt = 0; // The first part of the limit (offset)
        $num = 0; // The number of rows updated
        while ($num < $this->limit) {
            $rows = $queryBuilder
                ->select('*')->from($this->entity)
                ->setFirstResult($startAt)->setMaxResults($this->batchSize)
                ->orderBy($this->priKey)
                ->execute();

            // I need to read line by line if I have to update the table
            // to make sure I do update by update (slower but no other choice for now)
            foreach ($rows as $row) {
                // Call the right method according to the mode
                $this->{$this->updateMode[$this->mode]}($row);

                if ($callback !== null) {
                    $callback(++$num);
                }
                // Have to exit now as we have reached the max
                if ($num >= $this->limit) {
                    break 2;
                }
            }
            // Move the offset
            // Make sure the loop ends if we have nothing to process
            $num = $startAt += $this->batchSize;
        }
        // Run a final method if defined
        if ($this->mode === 'batch') {
            $this->loadDataInBatch('update');
        }

        $this->expression->evaluateExpressions(
            $this->configEntities[$this->entity]['post_actions']
        );
    }
719772
public function transaction(Closure $callback, $attempts = 1)
    {
        for ($currentAttempt = 1; $currentAttempt <= $attempts; ++$currentAttempt) {
            $this->beginTransaction();

            // We'll simply execute the given callback within a try / catch block and if we
            // catch any exception we can rollback this transaction so that none of this
            // gets actually persisted to a database or stored in a permanent fashion.
            try {
                return tap($callback($this), function ($result) {
                    $this->commit();
                });
            }

            // If we catch an exception we'll rollback this transaction and try again if we
            // are not out of attempts. If we are out of attempts we will just throw the
            // exception back out and let the developer handle an uncaught exceptions.
            catch (Exception $e) {
                $this->handleTransactionException(
                    $e,
                    $currentAttempt,
                    $attempts
                );
            } catch (Throwable $e) {
                $this->rollBack();

                throw $e;
            }
        }
    }