/*
 * Copyright (c) 2002-2010 JSCAPE
 * 1147 S. 53rd Pl., Mesa, Arizona, 85206, U.S.A.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * JSCAPE. ("Confidential Information").  You shall not disclose such
 * Confidential Information and shall use it only in accordance with
 *  the terms of the license agreement you entered into with JSCAPE.
 */

package com.jscape.util.net.connection;

import com.jscape.util.io.DataCodec;
import com.jscape.util.net.TransportAddress;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.jscape.util.Parameters.assertNotNull;

/**
 * @author Alex Usun
 */
public class AsyncMessageConnectionSyncRawBase<M>
        implements AsyncMessageConnection<M>,
                   MessageCodecConnection<M>,
                   Runnable {

    private final SyncMessageConnectionSyncRawBase<M> base;
    private final ExecutorService executorService;
    private final AtomicBoolean closed;
    private final AtomicBoolean eofReceived;
    private Listener<M> listener;

    public AsyncMessageConnectionSyncRawBase(SyncRawConnection base, DataCodec<M> messageCodec, ExecutorService executorService) {
        this(new SyncMessageConnectionSyncRawBase<M>(base, messageCodec), executorService);
    }

    public AsyncMessageConnectionSyncRawBase(SyncMessageConnectionSyncRawBase<M> base, ExecutorService executorService) {
        assertNotNull(base);
        this.base = base;

        assertNotNull(executorService);
        this.executorService = executorService;

        this.closed = new AtomicBoolean(false);
        this.eofReceived = new AtomicBoolean(false);
    }

    public long creationTime() {
        return this.base.creationTime();
    }

    public TransportAddress localAddress() {
        return this.base.localAddress();
    }

    public TransportAddress remoteAddress() {
        return this.base.remoteAddress();
    }

    @Override
    public Map<Object, Object> attributes() {
        return this.base.attributes();
    }

    public boolean closed() {
        return this.closed.get();
    }

    public void close() {
        if (this.closed.getAndSet(true)) {
            return;
        }

        this.base.close();

        if (this.listener != null) {
            this.listener.onClosed(this);
        }
    }

    public void start(Listener<M> listener)
            throws ConnectionException {
        assertNotNull(listener);
        this.listener = listener;

        this.executorService.submit(this);
    }

    public void write(M message)
            throws ConnectionException {
        this.base.write(message);
    }

    public void run() {
        try {
            while (!this.closed.get() && !this.eofReceived.get()) {
                readNextMessage();
            }
        } catch (Throwable e) {
            processError(e);
        }
    }

    public SyncRawConnection base() {
        return this.base.base();
    }

    public DataCodec<M> messageCodec() {
        return this.base.messageCodec();
    }

    public void messageCodec(DataCodec<M> value) {
        this.base.messageCodec(value);
    }

    @Override
    public String toString() {
        return this.base.toString();
    }

    private void readNextMessage()
            throws Exception {
        try {
            M message = this.base.read();
            this.listener.onMessage(this, message);
        } catch (ReadTimeoutException e) {
            if (!this.closed.get()) {
                this.listener.onIdle(this);
            }
        } catch (EOFException e) {
            if (!this.closed.get()) {
                this.eofReceived.set(true);
                this.listener.onEof(this);
            }
        }
    }

    private void processError(Throwable e) {
        if (!this.closed.get()) {
            this.listener.onError(this, e);
        }
    }

}
