/*
 * Copyright (c) 2002-2011 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.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 AsyncRawConnectionSyncRawBase
        implements AsyncRawConnection,
                   Runnable {

    private static final int EOF = -1;

    private final SyncRawConnection base;
    private final ExecutorService executorService;
    private final byte[] buffer;
    private final AtomicBoolean closed;
    private final AtomicBoolean eofReceived;
    private Listener listener;

    public AsyncRawConnectionSyncRawBase(SyncRawConnection base, ExecutorService executorService, int bufferSize) {
        assertNotNull(base);
        this.base = base;

        assertNotNull(executorService);
        this.executorService = executorService;

        this.buffer = new byte[bufferSize];
        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 listener)
            throws ConnectionException {
        assertNotNull(listener);
        this.listener = listener;

        this.executorService.submit(this);
    }

    public void write(byte[] data, int offset, int length)
            throws ConnectionException {
        this.base.write(data, offset, length);
    }

    public void write(byte data)
            throws ConnectionException {
        this.base.write(data);
    }

    public void flush()
            throws ConnectionException {
        this.base.flush();
    }

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

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

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

    protected void readNextData()
            throws Exception {
        try {
            int read = this.base.read(this.buffer, 0, this.buffer.length);

            if (read == EOF) {
                this.eofReceived.set(true);
                this.listener.onEof(this);
            } else {
                this.listener.onData(this, this.buffer, 0, read);
            }
        } 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);
            }
        }
    }

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

}
