@param Envelope $envelope

@throws ApiException If the DigiDoc envelope is not merged
768499
private function failIfNotMerged(Envelope $envelope)
    {
        if (!$this->tracker->has($envelope)) {
            throw ApiException::createNotMerged($envelope);
        }
    }
817652
public function send(\BearFramework\Emails\Email $email): void
    {
        $app = App::get();
        $email = clone($email);

        if ($this->hasEventListeners('beforeSendEmail')) {
            $eventDetails = new \BearFramework\Emails\BeforeSendEmailEventDetails($email);
            $this->dispatchEvent('beforeSendEmail', $eventDetails);
            if ($eventDetails->preventDefault) {
                return;
            }
        }

        if (empty($this->senders)) {
            throw new \Exception('No email senders added.');
        }

        foreach ($this->senders as $sender) {
            if (is_callable($sender)) {
                $sender = call_user_func($sender);
            }
            if (is_string($sender)) {
                $sender = new $sender();
            }
            if ($sender instanceof \BearFramework\Emails\ISender) {
                if ($sender->send($email)) {
                    if ($this->hasEventListeners('sendEmail')) {
                        $eventDetails = new \BearFramework\Emails\SendEmailEventDetails($email);
                        $this->dispatchEvent('sendEmail', $eventDetails);
                    }
                    return;
                }
            }
        }
        throw new \Exception('No email sender is capable of sending the email provided.');
    }
1074134
public void reply(MessageEnvelope envelope) {
		FbBotMillResponse response = createResponse(envelope);
		if (response != null) {
			// If the response is valid, replies to it.
			if (validate(response)) {
				FbBotMillNetworkController.postJsonMessage(response);
			}
		}
	}
557365
public function send($method, $uri, $body = null, array $query = [], array $headers = [], array $options = [])
    {
        $options = $this->enhanceOptions($body, $query, $headers, $options);

        try {
            $this->lastResponse = $this->client->request($method, $uri, $options);
        } catch (GuzzleException $e) {
            throw new ApiException($e->getMessage(), $e->getCode());
        }

        $this->validateResponse($this->lastResponse);

        return $this->lastResponse;
    }
1221000
protected function consume(Message $message, Envelope $envelope = null) : bool
    {
        if (!$envelope instanceof Job) {
            throw new InvalidException(sprintf("Invalid class '%s'", get_class($envelope)));
        }

        try {
            $job = $envelope->getCommand();

            $options = json_encode($job->getInput()->getOptions());
            $arguments = json_encode($job->getInput()->getArguments());

            if (!$this->output->isDebug()) {
                $arguments = strlen($arguments) > 50 ? substr($arguments, 0, 50) . '...' : $arguments;
                $options = strlen($options) > 50 ? substr($options, 0, 50) . '...' : $options;
            }

            $this->output->writeln(sprintf(
                "New job with options '%s' and arguments '%s'",
                $envelope->getClass(),
                $options,
                $arguments
            ), OutputInterface::VERBOSITY_VERY_VERBOSE);

            $job->setOutput($this->output);

            $return = $job->execute($job->input, $this->output) == 0;

            return $return;
        } catch (\Throwable $exception) {
            if (!$exception instanceof \Exception) {
                $exception = new FatalThrowableError($exception);
            }

            $this->output->writeError(sprintf(
                "%s on job '%s' with message '%s'",
                get_class($exception),
                $envelope->getClass(),
                $exception->getMessage()
            ));

            Application::writeException($exception, $this->output, ConsoleOutputInterface::VERBOSITY_VERBOSE);

            self::emit(new ErrorEvent($exception));

            if ($this->input->getOption('retry') && !$exception instanceof FailureException) {
                if ($this->input->getOption('retry') > count($envelope->getHistory())) {
                    if ($this->input->getOption('stop-on-error')) {
                        $this->exitCode = 1;
                    }

                    $this->queue()->publish($envelope->serialize());
                    throw new ManualRequeueException($this->input->getOption('stop-on-error') === 1);
                } else {
                    $errorMessage = sprintf(
                        "Max retries %s reach for %s on job '%s' with message '%s'",
                        $this->input->getOption('retry'),
                        get_class($exception),
                        $envelope->getClass(),
                        $exception->getMessage()
                    );

                    $this->output->writeError($errorMessage);
                    throw new FailureException($message, $errorMessage, $exception);
                }
            }

            if ($this->input->getOption('stop-on-error') && !$exception instanceof FailureException) {
                $this->exitCode = 1;
                throw $exception;
            }

            return true;
        }
    }
1020278
private function handleEvent(Envelope $envelope, Request $request, Event $event): Envelope
    {
        try {
            $this->pbjx->publish($event);
        } catch (\Throwable $e) {
            return $this->handleException($envelope, $request, $event, $e);
        }

        $envelope
            ->set('code', Code::OK)
            ->set('http_code', HttpCode::HTTP_ACCEPTED())
            ->set('message_ref', $event->generateMessageRef());

        if ($request->attributes->getBoolean('pbjx_console')) {
            $envelope->set('message', $event);
        }

        return $envelope;
    }
793631
private function isPbjxTokenOk(Envelope $envelope, Request $request, string $content): bool
    {
        if ($request->attributes->getBoolean('pbjx_console')) {
            // no tokens used on the console
            return true;
        }

        $curie = $request->attributes->get('pbjx_curie');
        foreach ($this->bypassTokenValidation as $pattern) {
            if ('all' === $pattern || $curie === $pattern) {
                return true;
            }

            if (preg_match('/' . trim($pattern, '/') . '/', $curie)) {
                return true;
            }
        }

        $token = $request->headers->get('x-pbjx-token');
        if (empty($token)) {
            $envelope
                ->set('code', Code::INVALID_ARGUMENT)
                ->set('http_code', HttpCode::HTTP_BAD_REQUEST())
                ->set('error_name', 'MissingPbjxToken')
                ->set('error_message', 'Missing x-pbjx-token header.');
            return false;
        }

        if ($request->headers->has('Authorization')) {
            $bearer = trim(str_ireplace('bearer ', '', $request->headers->get('Authorization')));
            $this->signer->addKey('bearer', $bearer);
        }

        if ($request->headers->has('x-pbjx-nonce')) {
            $this->signer->addKey('nonce', $request->headers->get('x-pbjx-nonce'));
        }

        try {
            $this->signer->validate($content, $request->getUri(), $token);
            return true;
        } catch (\Throwable $e) {
            $envelope
                ->set('code', Code::INVALID_ARGUMENT)
                ->set('http_code', HttpCode::HTTP_BAD_REQUEST())
                ->set('error_name', 'InvalidPbjxToken')
                ->set('error_message', $e->getMessage());
            return false;
        } finally {
            $this->signer->removeKey('bearer');
            $this->signer->removeKey('nonce');
        }
    }
767505
private function isContentTypeOk(Envelope $envelope, Request $request): bool
    {
        if ($request->query->has('callback') && $request->isMethod('GET')) {
            // jsonp request, don't enforce
            $request->attributes->set('_jsonp_enabled', true);
            return true;
        }

        $request->attributes->set('_jsonp_enabled', false);
        $contentType = $request->headers->get('Content-Type');

        if (0 === strpos($contentType, 'multipart/form-data') || 0 === strpos($contentType, 'application/json')) {
            return true;
        }

        $envelope
            ->set('code', Code::INVALID_ARGUMENT)
            ->set('http_code', HttpCode::HTTP_NOT_ACCEPTABLE())
            ->set('error_name', 'NotAcceptable')
            ->set(
                'error_message',
                sprintf(
                    'This service supports [application/json] or [multipart/form-data], you provided [%s].',
                    $contentType
                )
            );

        return false;
    }
947842
private void processEnvelope(Envelope envelope) {

        ActorScope scope = envelope.getScope();

        if (actorSystem.getTraceInterface() != null) {
            actorSystem.getTraceInterface().onEnvelopeDelivered(envelope);
        }

        long start = ActorTime.currentTime();
        if (scope.getActor() == null) {

            if (envelope.getMessage() == PoisonPill.INSTANCE) {
                // Not creating actor for PoisonPill
                return;
            }

            try {
                Actor actor = scope.getProps().create();
                actor.initActor(scope.getPath(), new ActorContext(scope), scope.getMailbox());
                ThreadDispatcher.pushDispatcher(actor.getDispatcher());
                try {
                    actor.preStart();
                } finally {
                    ThreadDispatcher.popDispatcher();
                }

                scope.onActorCreated(actor);
            } catch (Exception e) {
                e.printStackTrace();
                if (envelope.getSender() != null) {
                    envelope.getSender().send(new DeadLetter("Unable to create actor"));
                }
                return;
            }
        }

        try {
            if (envelope.getMessage() == StartActor.INSTANCE) {
                // Already created actor
            } else if (envelope.getMessage() == PoisonPill.INSTANCE) {
                ThreadDispatcher.pushDispatcher(scope.getActor().getDispatcher());
                try {
                    scope.getActor().postStop();
                } finally {
                    ThreadDispatcher.popDispatcher();
                }
                onActorDie(scope);
            } else {
                scope.getActor().handleMessage(envelope.getMessage(), envelope.getSender());
            }
        } catch (Exception e) {
            if (actorSystem.getTraceInterface() != null) {
                actorSystem.getTraceInterface().onActorDie(scope.getActorRef(), envelope, e);
            }
            ThreadDispatcher.pushDispatcher(scope.getActor().getDispatcher());
            try {
                scope.getActor().postStop();
            } finally {
                ThreadDispatcher.popDispatcher();
            }
            onActorDie(scope);
        } finally {
            if (actorSystem.getTraceInterface() != null) {
                actorSystem.getTraceInterface().onEnvelopeProcessed(envelope, ActorTime.currentTime() - start);
            }
        }
    }
1257022
void connectEnvelope(Envelope envelope) throws TransactionException {

        if (shutdown) {
            logger.warn(format("%s not connecting is shutdown.", toString()));
            return;
        }

        final AtomicBoolean retry = new AtomicBoolean(true); // make sure we only retry connection once for each connection attempt.

        ManagedChannel lmanagedChannel = managedChannel;

        if (lmanagedChannel == null || lmanagedChannel.isTerminated() || lmanagedChannel.isShutdown()) {

            lmanagedChannel = channelBuilder.build();
            managedChannel = lmanagedChannel;

        }

        try {

            DeliverGrpc.DeliverStub broadcast = DeliverGrpc.newStub(lmanagedChannel);

            // final DeliverResponse[] ret = new DeliverResponse[1];
            //   final List<DeliverResponse> retList = new ArrayList<>();
            final List<Throwable> throwableList = new ArrayList<>();
            final CountDownLatch finishLatch = new CountDownLatch(1);

            so = new StreamObserver<DeliverResponse>() {

                @Override
                public void onNext(DeliverResponse resp) {

                    // logger.info("Got Broadcast response: " + resp);
                    logger.trace(format("DeliverResponse %s resp status value:%d  status %s, typecase %s ",
                            PeerEventServiceClient.this.toString(), resp.getStatusValue(), resp.getStatus(), resp.getTypeCase()));

                    final DeliverResponse.TypeCase typeCase = resp.getTypeCase();

                    if (typeCase == STATUS) {

                        logger.debug(format("DeliverResponse  %s setting done.",
                                PeerEventServiceClient.this.toString()));

                        if (resp.getStatus() == Common.Status.SUCCESS) { // unlike you may think this only happens when all blocks are fetched.
                            peer.setLastConnectTime(System.currentTimeMillis());
                            peer.resetReconnectCount();
                        } else {

                            final long rec = peer.getReconnectCount();

                            PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("%s attempts %s Status returned failure code %d (%s) during peer service event registration",
                                    PeerEventServiceClient.this.toString(), rec, resp.getStatusValue(), resp.getStatus().name()));
                            peerEventingServiceException.setResponse(resp);
                            if (rec % 10 == 0) {
                                logger.warn(PeerEventServiceClient.this.toString() + " " + peerEventingServiceException.getMessage());
                            }

                            throwableList.add(peerEventingServiceException);
                        }

                    } else if (typeCase == FILTERED_BLOCK || typeCase == BLOCK) {
                        if (typeCase == BLOCK) {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getBlock().hashCode(), resp.getBlock().getHeader().getNumber()));
                        } else {
                            logger.trace(format("%s got event block hex hashcode: %016x, block number: %d",
                                    PeerEventServiceClient.this.toString(), resp.getFilteredBlock().hashCode(), resp.getFilteredBlock().getNumber()));
                        }

                        peer.setLastConnectTime(System.currentTimeMillis());
                        long reconnectCount = peer.getReconnectCount();
                        if (reconnectCount > 1) {
                            logger.info(format("%s reconnected after %d attempts on channel %s, peer %s, url %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, channelName, name, url));
                        }
                        peer.resetReconnectCount();

                        BlockEvent blockEvent = new BlockEvent(peer, resp);
                        peer.setLastBlockSeen(blockEvent);

                        channelEventQue.addBEvent(blockEvent);
                    } else {
                        logger.error(format("%s got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));

                        PeerEventingServiceException peerEventingServiceException = new PeerEventingServiceException(format("% got event block with unknown type: %s, %d",
                                PeerEventServiceClient.this.toString(), typeCase.name(), typeCase.getNumber()));
                        peerEventingServiceException.setResponse(resp);

                        throwableList.add(peerEventingServiceException);

                    }
                    finishLatch.countDown();

                }

                @Override
                public void onError(Throwable t) {
                    ManagedChannel llmanagedChannel = managedChannel;
                    if (llmanagedChannel != null) {
                        try {
                            llmanagedChannel.shutdownNow();
                        } catch (Exception e) {
                            logger.warn(format("Received error on %s, attempts %d. %s shut down of grpc channel.",
                                    PeerEventServiceClient.this.toString(), peer == null ? -1 : peer.getReconnectCount(), e.getMessage()), e);
                        }
                        managedChannel = null;
                    }
                    if (!shutdown) {
                        final long reconnectCount = peer.getReconnectCount();
                        if (PEER_EVENT_RECONNECTION_WARNING_RATE > 1 && reconnectCount % PEER_EVENT_RECONNECTION_WARNING_RATE == 1) {
                            logger.warn(format("Received error on  %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        } else {
                            logger.trace(format("Received error on %s, attempts %d. %s",
                                    PeerEventServiceClient.this.toString(), reconnectCount, t.getMessage()));

                        }

                        if (retry.getAndSet(false)) {
                            peer.reconnectPeerEventServiceClient(PeerEventServiceClient.this, t);
                        }

                    }
                    finishLatch.countDown();
                }

                @Override
                public void onCompleted() {
                    logger.debug(format("DeliverResponse onCompleted %s setting done.",
                            PeerEventServiceClient.this.toString()));
                    //            done = true;
                    //There should have been a done before this...
                    finishLatch.countDown();
                }
            };

            nso = filterBlock ? broadcast.deliverFiltered(so) : broadcast.deliver(so);

            nso.onNext(envelope);

            // try {
            if (!finishLatch.await(peerEventRegistrationWaitTimeMilliSecs, TimeUnit.MILLISECONDS)) {
                PeerEventingServiceException ex = new PeerEventingServiceException(format(
                        "Channel %s connect time exceeded for peer eventing service %s, timed out at %d ms.", channelName, name, peerEventRegistrationWaitTimeMilliSecs));
                ex.setTimedOut(peerEventRegistrationWaitTimeMilliSecs);
                logger.warn(toString() + " " + ex.getMessage());
                throwableList.add(0, ex);

            }
            logger.trace(toString() + " done waiting for reply!");

            if (!throwableList.isEmpty()) {
                ManagedChannel llmanagedChannel = managedChannel;
                if (llmanagedChannel != null) {
                    llmanagedChannel.shutdownNow();
                    managedChannel = null;
                }
                Throwable throwable = throwableList.get(0);
                if (retry.getAndSet(false)) {
                    peer.reconnectPeerEventServiceClient(this, throwable);
                }

            }

        } catch (InterruptedException e) {
            ManagedChannel llmanagedChannel = managedChannel;
            if (llmanagedChannel != null) {
                llmanagedChannel.shutdownNow();
                managedChannel = null;
            }
            logger.error(toString() + " error message: " + e.getMessage(), e); // not likely

            if (retry.getAndSet(false)) {
                peer.reconnectPeerEventServiceClient(this, e);
            }

        } finally {
            if (null != nso) {

                try {
                    nso.onCompleted();
                } catch (Exception e) {  //Best effort only report on debug
                    logger.debug(format("Exception completing connect with %s %s",
                            toString(), e.getMessage()), e);
                }

            }
        }
    }
171438
public function setApiVersion($apiVersion)
    {
        if ( ! in_array($apiVersion, self::$supportedApiVersions)) {
            throw new ApiVersionNotSupported(sprintf('API version %d is not supported.', $apiVersion));
        }
        $this->apiVersion = $apiVersion;
    }
1091956
public function invoke(Envelope $envelope, Queue $queue)
    {
        try {
            $this->dispatcher->dispatch(BernardEvents::INVOKE, new EnvelopeEvent($envelope, $queue));

            $receiver = $this->router->route($envelope);
            $receiver->receive($envelope->getMessage());

            // We successfully processed the message.
            $queue->acknowledge($envelope);

            $this->dispatcher->dispatch(BernardEvents::ACKNOWLEDGE, new EnvelopeEvent($envelope, $queue));
        } catch (\Throwable $error) {
            $this->rejectDispatch($error, $envelope, $queue);
        } catch (\Exception $exception) {
            $this->rejectDispatch($exception, $envelope, $queue);
        }
    }
227847
protected function doSendEvent(Event $event): void
    {
        $envelope = new TransportEnvelope($event, 'json');
        $this->client->putRecord([
            'StreamName'   => $this->router->forEvent($event),
            'PartitionKey' => $this->router->partitionForEvent($event),
            'Data'         => $envelope->toString(),
        ]);
    }
1223920
public void setEnvelopeFrom(String address) throws AddressException {
		if(StringUtils.isEmpty(address)) {
			this.envelopeFrom = address;
		} else {
			InternetAddress tmp = toInternetAddress(address, null);
			this.envelopeFrom = tmp.getAddress();
		}
	}
3060
public static function fromString(string $envelope): self
    {
        $envelope = json_decode($envelope, true);
        if (!is_array($envelope)) {
            throw new \InvalidArgumentException('Envelope is invalid. ' . json_last_error_msg());
        }

        $serializer = isset($envelope['serializer']) ? $envelope['serializer'] : 'php';
        $isReplay = isset($envelope['is_replay']) ? filter_var($envelope['is_replay'], FILTER_VALIDATE_BOOLEAN) : false;
        $message = self::getSerializer($serializer)->deserialize(isset($envelope['message']) ? $envelope['message'] : '');

        /*
         * don't attempt to set replay if it's the php serializer because it already knows
         * if it's a replay as the entire object is serialized with its internal state.
         *
         * this is a PHP feature only, all other serializers have no notion of the
         * language specific in memory object.  we use php serializer because it's
         * very fast to un/de-serialize the objects.
         */
        if ($isReplay && 'php' !== $serializer) {
            $message->isReplay(true);
        }

        return new self($message, $serializer);
    }
748354
public void setEnvelope(Envelope env) {
		if (!m_description.equals(env.getDescription()))
			throw new IllegalArgumentException();

		// m_envelope = (Envelope) env.clone();
		m_envelope = (Envelope) env.createInstance();
		env.copyTo(m_envelope);
		_setDirtyFlag(DirtyFlags.DirtyIntervals, false);
	}
232806
private function rejectDispatch($exception, Envelope $envelope, Queue $queue)
    {
        // Make sure the exception is not interfering.
        // Previously failing jobs handling have been moved to a middleware.
        //
        // Emit an event to let others log that exception
        $this->dispatcher->dispatch(BernardEvents::REJECT, new RejectEnvelopeEvent($envelope, $queue, $exception));

        if ($this->options['stop-on-error']) {
            throw $exception;
        }
    }