/**
 * Copyright 2024 Beijing Baolande Software Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * Runtime Library Exception to the Apache 2.0 License:
 *
 * As an exception, if you use this Software to compile your source code and
 * portions of this Software are embedded into the binary product as a result,
 * you may redistribute such product without providing attribution as would
 * otherwise be required by Sections 4(a), 4(b) and 4(d) of the License.
 */

package redis_sdk.client

public open class RedisSubscriberBuilder <: RedisClientBuilderBase {
    public static func builder(): RedisSubscriberBuilder {
        return RedisSubscriberBuilder()
    }

    public static func builder(clientConfig: RedisClientConfig): RedisSubscriberBuilder {
        return RedisSubscriberBuilder(clientConfig)
    }

    public static func buildRedisSubscriber(host: String, port: UInt16): RedisSubscriber {
        let clientBuilder = builder()
        return clientBuilder.host(host).port(port).build()
    }

    public static func buildRedisSubscriber(host: String, port: UInt16, password: String): RedisSubscriber {
        let clientBuilder = builder()
        return clientBuilder.host(host).port(port).password(password).build()
    }

    public static func buildRedisSubscriber(host: String, port: UInt16, user: String, password: String): RedisSubscriber {
        let clientBuilder = builder()
        return clientBuilder.host(host).port(port).user(user).password(password).build()
    }

    let redisClientConfig: RedisClientConfig

    /**
     * Protected Constructor
     */
    protected init() {
        this(RedisClientConfig())
    }

    /**
     * Protected Constructor
     */
    protected init(clientConfig: RedisClientConfig) {
        super(clientConfig)
        this.redisClientConfig = clientConfig
    }

    public func host(host: String) {
        redisClientConfig.host = host
        return this
    }

    public func port(port: UInt16) {
        redisClientConfig.port = port
        return this
    }

    public func checkSubscribeMessageInterval(checkSubscribeMessageInterval: Duration) {
        redisClientConfig.checkSubscribeMessageInterval = checkSubscribeMessageInterval
        return this
    }

    public open func build(): RedisSubscriber {
        let tcpEndpoint = RedisClientBuilder.buildClientEndPoint(this.redisClientConfig)
        let subscribeHandler = RedisSubscriberHandler.getInstance()
        configEndpoint(tcpEndpoint, subscribeHandler)

        let commandExecutor = DefaultCommandExecutor(tcpEndpoint)
        let redisSubscriber = RedisSubscriber(commandExecutor, subscribeHandler)
        redisSubscriber.setCheckSubscribeMessageInterval(redisClientConfig.checkSubscribeMessageInterval)

        let handshakeInfo = RedisClientBuilder.getHandShakeInfo(this.redisClientConfig)
        let clientInitializer: RedisClientInitializer = RedisClientInitializer(
            commandExecutor.getRespVersionAware(),
            handshakeInfo,
            redisClientConfig.waitResponseTimeout
        )
        tcpEndpoint.setConnectionInitializer(clientInitializer)

        tcpEndpoint.start()
        return redisSubscriber
    }

    public static func configEndpoint(tcpEndpoint: ClientTcpEndpoint, subscribeHandler: RedisSubscriberHandler) {
        // RedisCodec
        let encoder = RedisCommandToByteEncoder()
        let decoder = ByteToRedisMessageDecoder()
        let redisCodec = RedisCodec(encoder, decoder)

        tcpEndpoint.setMessageCompletedHandler(decoder)

        // RedisSubscriberHandler
        tcpEndpoint.addFilter(redisCodec)
        tcpEndpoint.addFilter(subscribeHandler)
    }
}