Does a streaming server-sent event response which will loop and execute
the specified callback indefinitely and update the client only when
needed.

@param $callback
@param string $eventId
@param int $retry
@return Response
423678
protected function sseStreaming($callback, $eventId = 'stream', $sleep = 1): Response
    {
        return ResponseFactory::getInstance()->buildStreamingSse($callback, $eventId, $sleep);
    }
1083310
private function createEventListenerForStream($callback)
    {
        $logger = $this->getLogger();

        $function = function (MessageEvent $event) use ($callback, $logger) {
            $protocol = $event->getProtocolMessage();

            if ($protocol === null) {
                $logger->debug('Incompatable message: '.$event->getMessage());

                return;
            }

            $routing = $event->getMessage()->getRoutingInformation();

            call_user_func($callback, $protocol, $routing);

            return;
        };

        return $function;
    }
369296
public function withCallback($successCallback, $failedCallback = null, $callbackMode = self::CALLBACK_MODE_HTTP)
    {
        $this->callback = $successCallback;
        $this->failedCallback = $failedCallback;
        $this->callbackMode = $callbackMode;

        return $this;
    }
13082
public function update($id, EventCallbackUrl $eventCallbackUrl)
    {
        $response = $this->mailjet->put(Resources::$Eventcallbackurl, ['id' => $id, 'body' => $eventCallbackUrl->format()]);
        if (!$response->success()) {
            $this->throwError("EventCallbackUrlService:update() failed", $response);
        }

        return $response->getData();
    }
1275268
public function send(RequestInterface $request): ResponseInterface
    {
        $this->emit('beforeRequest', [$request]);

        $retryCount = 0;
        $redirects = 0;

        do {
            $doRedirect = false;
            $retry = false;

            try {
                $response = $this->doRequest($request);

                $code = $response->getStatus();

                // We are doing in-PHP redirects, because curl's
                // FOLLOW_LOCATION throws errors when PHP is configured with
                // open_basedir.
                //
                // https://github.com/fruux/sabre-http/issues/12
                if ($redirects < $this->maxRedirects && in_array($code, [301, 302, 307, 308])) {
                    $oldLocation = $request->getUrl();

                    // Creating a new instance of the request object.
                    $request = clone $request;

                    // Setting the new location
                    $request->setUrl(Uri\resolve(
                        $oldLocation,
                        $response->getHeader('Location')
                    ));

                    $doRedirect = true;
                    ++$redirects;
                }

                // This was a HTTP error
                if ($code >= 400) {
                    $this->emit('error', [$request, $response, &$retry, $retryCount]);
                    $this->emit('error:'.$code, [$request, $response, &$retry, $retryCount]);
                }
            } catch (ClientException $e) {
                $this->emit('exception', [$request, $e, &$retry, $retryCount]);

                // If retry was still set to false, it means no event handler
                // dealt with the problem. In this case we just re-throw the
                // exception.
                if (!$retry) {
                    throw $e;
                }
            }

            if ($retry) {
                ++$retryCount;
            }
        } while ($retry || $doRedirect);

        $this->emit('afterRequest', [$request, $response]);

        if ($this->throwExceptions && $code >= 400) {
            throw new ClientHttpException($response);
        }

        return $response;
    }
706015
public function then(callable $callback, callable $failure = null)
    {
        try {
            return (new DurationLimiter(
                $this->connection, $this->name, $this->maxLocks, $this->decay
            ))->block($this->timeout, $callback);
        } catch (LimiterTimeoutException $e) {
            if ($failure) {
                return $failure($e);
            }

            throw $e;
        }
    }
823897
private function process($retry = 0)
    {
        $ch = curl_init();

        $option = array_merge($this->option, $this->custom);
        foreach($option as $key => $val) {
            if (is_string($key)) {
                $key = constant(strtoupper($key));
            }
            curl_setopt($ch, $key, $val);
        }

        if ($this->post) {
            curl_setopt($ch, CURLOPT_POST, true);
            curl_setopt($ch, CURLOPT_POSTFIELDS, $this->convert($this->post));
        }

        $this->data = (string) curl_exec($ch);
        $this->info = curl_getinfo($ch);
        $this->error = curl_errno($ch);
        $this->message = $this->error ? curl_error($ch) : '';

        curl_close($ch);

        if ($this->error && $retry < $this->retry) {
            $this->process($retry + 1);
        }

        $this->post     = array();
        $this->retry    = 0;

        return $this;
    }
140063
public function onRequestError(Event $event)
    {
        // if token validity expired, re-request with a new token.
        if (in_array($event['response']->getStatusCode(), array(401, 403))) {
            /** @var \Guzzle\Http\Message\Request $request */
            $request = $event['request'];

            // if this is the token-url, stop now because we won't be able to fetch a token
            if ($request->getUrl() === $request->getClient()->getTokenUrl()) {
                return;
            }

            // see if we have retries left
            if ($request->hasHeader('X-Auth-Retries')) {
                $headerValues = $request->getHeader('X-Auth-Retries')->toArray();
                $retriesValue = array_shift($headerValues);
            }

            $retries = $request->hasHeader('X-Auth-Retries') ? $retriesValue : 1;
            if ($retries < 1) {
                if ($this->logger) {
                    $this->logger->addError('Keystone request failed, no more retries left');
                }

                return;
            }

            if ($this->logger) {
                $this->logger->addDebug('Token expired, fetching a new one');
            }

            // set new token in client
            $client = $request->getClient();
            $this->resetToken($client);

            // clone request and update token header
            $newRequest = clone $request;
            $newRequest->setHeader('X-Auth-Token', $client->getToken()->getId());
            $newRequest->setHeader('X-Auth-Retries', --$retries);
            $newResponse = $newRequest->send();

            // Set the response object of the request without firing more events
            $event['response'] = $newResponse;

            // Stop other events from firing when you override 401 responses
            $event->stopPropagation();
        }
    }
775746
function() {
        // Send request
        var sendRequest = function() {
          debug('Calling: %s, retry: %s', entry.name, attempts - 1);
          // Make request and handle response or error
          return makeRequest(
            that,
            entry.method,
            url,
            payload,
            query
          ).then(function(res) {
            // If request was successful, accept the result
            debug('Success calling: %s, (%s retries)',
              entry.name, attempts - 1);
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure([entry.name, 'success'], d[0] * 1000 + d[1] / 1000000);
              monitor.count([entry.name, 'success']);
            }
            if (!_.includes(res.headers['content-type'], 'application/json') || !res.body) {
              debug('Empty response from server: call: %s, method: %s', entry.name, entry.method);
              return undefined;
            }
            return res.body;
          }, function(err) {
            // If we got a response we read the error code from the response
            var res = err.response;
            if (res) {
              // Decide if we should retry
              if (attempts <= that._options.retries &&
                  res.status >= 500 &&  // Check if it's a 5xx error
                  res.status < 600) {
                debug('Error calling: %s now retrying, info: %j',
                  entry.name, res.body);
                return retryRequest();
              }
              // If not retrying, construct error object and reject
              debug('Error calling: %s NOT retrying!, info: %j',
                entry.name, res.body);
              var message = 'Unknown Server Error';
              if (res.status === 401) {
                message = 'Authentication Error';
              }
              if (res.status === 500) {
                message = 'Internal Server Error';
              }
              err = new Error(res.body.message || message);
              err.body = res.body;
              err.code = res.body.code || 'UnknownError';
              err.statusCode = res.status;
              if (monitor) {
                var d = process.hrtime(start);

                var state = 'client-error';
                if (res.statusCode >= 500) {
                  state = 'server-error';
                }
                monitor.measure([entry.name, state], d[0] * 1000 + d[1] / 1000000);
                monitor.count([entry.name, state]);
              }
              throw err;
            }

            // Decide if we should retry
            if (attempts <= that._options.retries) {
              debug('Request error calling %s (retrying), err: %s, JSON: %s',
                entry.name, err, err);
              return retryRequest();
            }
            debug('Request error calling %s NOT retrying!, err: %s, JSON: %s',
              entry.name, err, err);
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure([entry.name, 'connection-error'], d[0] * 1000 + d[1] / 1000000);
              monitor.count([entry.name, 'connection-error']);
            }
            throw err;
          });
        };

        // Increment attempt count, but track how many we had before.
        attempts += 1;

        // If this is the first retry, ie we haven't retried yet, we make the
        // request immediately
        if (attempts === 1) {
          return sendRequest();
        } else {
          var delay;
          // First request is attempt = 1, so attempt = 2 is the first retry
          // we subtract one to get exponents: 1, 2, 3, 4, 5, ...
          delay = Math.pow(2, attempts - 1) * that._options.delayFactor;
          // Apply randomization factor
          var rf = that._options.randomizationFactor;
          delay *= Math.random() * 2 * rf + 1 - rf;
          // Always limit with a maximum delay
          delay = Math.min(delay, that._options.maxDelay);
          // Sleep then send the request
          return new Promise(function(accept) {
            setTimeout(accept, delay);
          }).then(function() {
            return sendRequest();
          });
        }
      }
1168413
public function listen($event, $callback)
    {
        if (!isset($this->listeners[$event])) {
            $this->listeners[$event] = [];
        }

        $this->listeners[$event][] = $callback;
    }
201006
private function tryRequest(
        callable $callable,
        ?Retry $retry
    ): array {
        $tries = $retry instanceof Retry
            ? $retry->getRetries()
            : 0;

        while (true) {
            try {
                return $callable();
            } catch (\Exception $e) {
                if ($tries-- <= 0) {
                    throw $e;
                }

                usleep($retry->getMicrosecondsBetweenRetries());
            }
        }
    }
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;
            }
        }
    }
445224
public function addRetry(Retry $retry)
    {
        $this->retries[$retry->getUrl().'~~'.$retry->getMethod()] = $retry;
    }
233779
public function block($timeout, $callback = null)
    {
        $starting = time();

        $id = Str::random(20);

        while (! $slot = $this->acquire($id)) {
            if (time() - $timeout >= $starting) {
                throw new LimiterTimeoutException;
            }

            usleep(250 * 1000);
        }

        if (is_callable($callback)) {
            try {
                return tap($callback(), function () use ($slot, $id) {
                    $this->release($slot, $id);
                });
            } catch (Exception $exception) {
                $this->release($slot, $id);

                throw $exception;
            }
        }

        return true;
    }
571875
public function setCallback(callable $callback): EventObserver
    {
        $this->callback = $callback;
        $this->id = $this->getId($callback);

        return $this;
    }
187471
@Override
    public void executeEventCallback(Runnable callback) {
        if (!nodeEngine.isRunning()) {
            return;
        }
        try {
            eventExecutor.execute(callback);
        } catch (RejectedExecutionException e) {
            rejectedCount.inc();

            if (eventExecutor.isLive()) {
                logFailure("EventQueue overloaded! Failed to execute event callback: %s", callback);
            }
        }
    }
531502
function emit(retry) {
    if (self._interrupted) {
      // The request's callback will already have been called.
      return;
    }

    var hreq = self._createHandshakeRequest(adapter, !retry);

    var writable = self._writableFactory.call(self, function (err, readable) {
      if (err) {
        cb(err);
        return;
      }
      readable.on('data', function (obj) {
        debug('received response %s', obj.id);
        // We don't check that the prefix matches since the ID likely hasn't
        // been propagated to the response (see default stateless codec).
        var buf = Buffer.concat(obj.payload);
        try {
          var parts = readHead(HANDSHAKE_RESPONSE_TYPE, buf);
          var hres = parts.head;
          if (hres.serverHash) {
            adapter = self._getAdapter(hres);
          }
        } catch (err) {
          cb(err);
          return;
        }
        self.emit('handshake', hreq, hres);
        if (hres.match === 'NONE') {
          process.nextTick(function() { emit(true); });
          return;
        }
        // Change the default adapter.
        self._adapter = adapter;
        cb(null, parts.tail, adapter);
      });
    });

    writable.write({
      id: id,
      payload: [HANDSHAKE_REQUEST_TYPE.toBuffer(hreq), reqBuf]
    });
    if (self._endWritable) {
      writable.end();
    }
  }