"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const logger_service_1 = require("@nestjs/common/services/logger.service");
const load_package_util_1 = require("@nestjs/common/utils/load-package.util");
const shared_utils_1 = require("@nestjs/common/utils/shared.utils");
const rxjs_1 = require("rxjs");
const constants_1 = require("../constants");
const invalid_grpc_package_exception_1 = require("../errors/invalid-grpc-package.exception");
const invalid_grpc_service_exception_1 = require("../errors/invalid-grpc-service.exception");
const invalid_proto_definition_exception_1 = require("../errors/invalid-proto-definition.exception");
const client_proxy_1 = require("./client-proxy");
const constants_2 = require("./constants");
let grpcPackage = {};
let grpcProtoLoaderPackage = {};
class ClientGrpcProxy extends client_proxy_1.ClientProxy {
    constructor(options) {
        super();
        this.options = options;
        this.logger = new logger_service_1.Logger(client_proxy_1.ClientProxy.name);
        this.clients = new Map();
        this.grpcClients = [];
        this.url = this.getOptionsProp(options, 'url') || constants_1.GRPC_DEFAULT_URL;
        const protoLoader = this.getOptionsProp(options, 'protoLoader') || constants_1.GRPC_DEFAULT_PROTO_LOADER;
        grpcPackage = load_package_util_1.loadPackage('grpc', ClientGrpcProxy.name, () => require('grpc'));
        grpcProtoLoaderPackage = load_package_util_1.loadPackage(protoLoader, ClientGrpcProxy.name);
        this.grpcClients = this.createClients();
    }
    getService(name) {
        const grpcClient = this.createClientByServiceName(name);
        const clientRef = this.getClient(name);
        if (!clientRef) {
            throw new invalid_grpc_service_exception_1.InvalidGrpcServiceException();
        }
        const protoMethods = Object.keys(clientRef[name].prototype);
        const grpcService = {};
        protoMethods.forEach(m => {
            const key = m[0].toLowerCase() + m.slice(1, m.length);
            grpcService[key] = this.createServiceMethod(grpcClient, m);
        });
        return grpcService;
    }
    getClientByServiceName(name) {
        return this.clients.get(name) || this.createClientByServiceName(name);
    }
    createClientByServiceName(name) {
        const clientRef = this.getClient(name);
        if (!clientRef) {
            throw new invalid_grpc_service_exception_1.InvalidGrpcServiceException();
        }
        const maxSendMessageLengthKey = 'grpc.max_send_message_length';
        const maxReceiveMessageLengthKey = 'grpc.max_receive_message_length';
        const maxMessageLengthOptions = {
            [maxSendMessageLengthKey]: this.getOptionsProp(this.options, 'maxSendMessageLength', constants_1.GRPC_DEFAULT_MAX_SEND_MESSAGE_LENGTH),
            [maxReceiveMessageLengthKey]: this.getOptionsProp(this.options, 'maxReceiveMessageLength', constants_1.GRPC_DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH),
        };
        const maxMetadataSize = this.getOptionsProp(this.options, 'maxMetadataSize', -1);
        if (maxMetadataSize > 0) {
            maxMessageLengthOptions['grpc.max_metadata_size'] = maxMetadataSize;
        }
        const keepaliveOptions = this.getKeepaliveOptions();
        const options = Object.assign(Object.assign({}, maxMessageLengthOptions), keepaliveOptions);
        const credentials = this.options.credentials || grpcPackage.credentials.createInsecure();
        const grpcClient = new clientRef[name](this.url, credentials, options);
        this.clients.set(name, grpcClient);
        return grpcClient;
    }
    getKeepaliveOptions() {
        if (!shared_utils_1.isObject(this.options.keepalive)) {
            return {};
        }
        const keepaliveKeys = {
            keepaliveTimeMs: 'grpc.keepalive_time_ms',
            keepaliveTimeoutMs: 'grpc.keepalive_timeout_ms',
            keepalivePermitWithoutCalls: 'grpc.keepalive_permit_without_calls',
            http2MaxPingsWithoutData: 'grpc.http2.max_pings_without_data',
            http2MinTimeBetweenPingsMs: 'grpc.http2.min_time_between_pings_ms',
            http2MinPingIntervalWithoutDataMs: 'grpc.http2.min_ping_interval_without_data_ms',
            http2MaxPingStrikes: 'grpc.http2.max_ping_strikes',
        };
        const keepaliveOptions = {};
        for (const [optionKey, optionValue] of Object.entries(this.options.keepalive)) {
            const key = keepaliveKeys[optionKey];
            if (!key) {
                continue;
            }
            keepaliveOptions[key] = optionValue;
        }
        return keepaliveOptions;
    }
    createServiceMethod(client, methodName) {
        return client[methodName].responseStream
            ? this.createStreamServiceMethod(client, methodName)
            : this.createUnaryServiceMethod(client, methodName);
    }
    createStreamServiceMethod(client, methodName) {
        return (...args) => {
            const isRequestStream = client[methodName].requestStream;
            const stream = new rxjs_1.Observable(observer => {
                let isClientCanceled = false;
                let upstreamSubscription;
                const upstreamSubjectOrData = args[0];
                const maybeMetadata = args[1];
                const isUpstreamSubject = upstreamSubjectOrData && shared_utils_1.isFunction(upstreamSubjectOrData.subscribe);
                const call = isRequestStream && isUpstreamSubject
                    ? client[methodName](maybeMetadata)
                    : client[methodName](...args);
                if (isRequestStream && isUpstreamSubject) {
                    upstreamSubscription = upstreamSubjectOrData.subscribe((val) => call.write(val), (err) => call.emit('error', err), () => call.end());
                }
                call.on('data', (data) => observer.next(data));
                call.on('error', (error) => {
                    if (error.details === constants_2.GRPC_CANCELLED) {
                        call.destroy();
                        if (isClientCanceled) {
                            return;
                        }
                    }
                    observer.error(error);
                });
                call.on('end', () => {
                    if (upstreamSubscription) {
                        upstreamSubscription.unsubscribe();
                        upstreamSubscription = null;
                    }
                    call.removeAllListeners();
                    observer.complete();
                });
                return () => {
                    if (upstreamSubscription) {
                        upstreamSubscription.unsubscribe();
                        upstreamSubscription = null;
                    }
                    if (call.finished) {
                        return undefined;
                    }
                    isClientCanceled = true;
                    call.cancel();
                };
            });
            return stream;
        };
    }
    createUnaryServiceMethod(client, methodName) {
        return (...args) => {
            const isRequestStream = client[methodName].requestStream;
            const upstreamSubjectOrData = args[0];
            const isUpstreamSubject = upstreamSubjectOrData && shared_utils_1.isFunction(upstreamSubjectOrData.subscribe);
            if (isRequestStream && isUpstreamSubject) {
                return new rxjs_1.Observable(observer => {
                    const callArgs = [
                        (error, data) => {
                            if (error) {
                                return observer.error(error);
                            }
                            observer.next(data);
                            observer.complete();
                        },
                    ];
                    const maybeMetadata = args[1];
                    if (maybeMetadata) {
                        callArgs.unshift(maybeMetadata);
                    }
                    const call = client[methodName](...callArgs);
                    upstreamSubjectOrData.subscribe((val) => call.write(val), (err) => call.emit('error', err), () => call.end());
                });
            }
            return new rxjs_1.Observable(observer => {
                client[methodName](...args, (error, data) => {
                    if (error) {
                        return observer.error(error);
                    }
                    observer.next(data);
                    observer.complete();
                });
            });
        };
    }
    createClients() {
        const grpcContext = this.loadProto();
        const packageOption = this.getOptionsProp(this.options, 'package');
        const grpcPackages = [];
        const packageNames = Array.isArray(packageOption)
            ? packageOption
            : [packageOption];
        for (const packageName of packageNames) {
            const grpcPkg = this.lookupPackage(grpcContext, packageName);
            if (!grpcPkg) {
                const invalidPackageError = new invalid_grpc_package_exception_1.InvalidGrpcPackageException();
                this.logger.error(invalidPackageError.message, invalidPackageError.stack);
                throw invalidPackageError;
            }
            grpcPackages.push(grpcPkg);
        }
        return grpcPackages;
    }
    loadProto() {
        try {
            const file = this.getOptionsProp(this.options, 'protoPath');
            const loader = this.getOptionsProp(this.options, 'loader');
            const packageDefinition = grpcProtoLoaderPackage.loadSync(file, loader);
            const packageObject = grpcPackage.loadPackageDefinition(packageDefinition);
            return packageObject;
        }
        catch (err) {
            const invalidProtoError = new invalid_proto_definition_exception_1.InvalidProtoDefinitionException();
            const message = err && err.message ? err.message : invalidProtoError.message;
            this.logger.error(message, invalidProtoError.stack);
            throw invalidProtoError;
        }
    }
    lookupPackage(root, packageName) {
        /** Reference: https://github.com/kondi/rxjs-grpc */
        let pkg = root;
        for (const name of packageName.split(/\./)) {
            pkg = pkg[name];
        }
        return pkg;
    }
    close() {
        this.grpcClients.forEach(client => client.close());
        this.grpcClients = [];
    }
    async connect() {
        throw new Error('The "connect()" method is not supported in gRPC mode.');
    }
    send(pattern, data) {
        throw new Error('Method is not supported in gRPC mode. Use ClientGrpc instead (learn more in the documentation).');
    }
    getClient(name) {
        return this.grpcClients.find(client => client.hasOwnProperty(name));
    }
    publish(packet, callback) {
        throw new Error('Method is not supported in gRPC mode. Use ClientGrpc instead (learn more in the documentation).');
    }
    async dispatchEvent(packet) {
        throw new Error('Method is not supported in gRPC mode. Use ClientGrpc instead (learn more in the documentation).');
    }
}
exports.ClientGrpcProxy = ClientGrpcProxy;