/*
 * Copyright Avaya Inc., All Rights Reserved.
 *
 * THIS IS UNPUBLISHED PROPRIETARY SOURCE CODE OF Avaya Inc.
 *
 * The copyright notice above does not evidence any actual or intended
 * publication of such source code.
 *
 * Some third-party source code components may have been modified from their
 * original versions by Avaya Inc.
 *
 * The modifications are Copyright Avaya Inc., All Rights Reserved.
 */

package com.avaya.oceanalytics.openinterface.websocketclient.websocket;

import com.avaya.oceanalytics.openinterface.websocketclient.subscription.HeartbeatGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Realtime data provider based on websocket.
 * 
 * @author seprokof
 *
 */
@Component
@Scope(ConfigurableBeanFactory.SCOPE_SINGLETON)
public class WebSocketRtdProvider implements RealtimeDataProvider {
    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketRtdProvider.class);

    private static final String SSL_CONTEXT_KEY = "org.apache.tomcat.websocket.SSL_CONTEXT";
    private static final int CONNECTION_TIME_LIMIT = 10;
    public static final int DELAY = 5 * 1000; // 5 seconds delay before task is executed
    public static final int PERIOD = 8 * 1000; // 8 seconds between successive task execution

    private final StandardWebSocketClient socket;
    private WebSocketSession session;

    private HeartbeatGenerator heartbeatGenerator;
    private Timer timer;
    private final AtomicBoolean isHeartbeatOn = new AtomicBoolean(false);


    @Autowired
    public WebSocketRtdProvider(SSLContext sslContext) {
        this.socket = new StandardWebSocketClient();
        Map<String, Object> userProperties = new HashMap<>();
        userProperties.put(SSL_CONTEXT_KEY, sslContext);
        this.socket.setUserProperties(userProperties);
    }

    @Override
    public void connect(String endpointUri, RealtimeDataHandler rtdHandler) {
        DelegatingWebsocketHandler webSocketHandler = new DelegatingWebsocketHandler();
        webSocketHandler.setRtdHandler(rtdHandler);
        ListenableFuture<WebSocketSession> webSessionFuture = socket.doHandshake(webSocketHandler, endpointUri);
        try {
            session = webSessionFuture.get(CONNECTION_TIME_LIMIT, TimeUnit.MINUTES);
            startHeartbeat();
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            throw new RealtimeDataProviderException("Unable to establish websocket connection", e);
        }
    }

    @Override
    public void startHeartbeat(){
        if (!isHeartbeatOn.get()) {
            heartbeatGenerator = new HeartbeatGenerator();
            heartbeatGenerator.setSession(session);
            timer = new Timer(true);
            timer.scheduleAtFixedRate(heartbeatGenerator, DELAY, PERIOD);
            isHeartbeatOn.set(true);
            LOGGER.info("Starting to send heartbeats to the WebSocket.");
        }
    }

    @Override
    public void stopHeartbeat(){
        if (isHeartbeatOn.get()) {
            timer.cancel();
            isHeartbeatOn.set(false);
            LOGGER.info("Stopping heartbeats to the WebSocket.");
        }
    }

    @Override
    public void send(String message) {
        if (session == null) {
            throw new RealtimeDataProviderException("Websocket is not connected");
        }
        try {
            synchronized (session) {
                session.sendMessage(new TextMessage(message));
            }
        } catch (IOException e) {
            throw new RealtimeDataProviderException("Unable to send message to websocket", e);
        }
    }

    @Override
    public void disconnect() {
        if (session != null) {
            try {
                stopHeartbeat();
                session.close(CloseStatus.NORMAL);
            } catch (IOException e) {
                LOGGER.error("Error while disconnecting the websocket", e);
            }
        }
    }

    @Override
    public boolean isAlive() {
        if(session == null) {
            return false;
        }
        return session.isOpen();
    }
}
