/*
 * 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.io.*;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.jscape.util.Parameters.assertNotNull;
import static com.jscape.util.net.connection.Connection.ConnectionException.wrap;

/**
 * @author Alex Usun
 */
public class SyncMessageConnectionSyncRawBase<M>
        implements SyncMessageConnection<M>,
                   MessageCodecConnection<M> {

    private final SyncRawConnection base;
    private DataCodec<M> messageCodec;
    private final OutputStream out;
    private final InputStream in;
    private final AtomicBoolean closed;

    public SyncMessageConnectionSyncRawBase(SyncRawConnection base, DataCodec<M> messageCodec) {
        assertNotNull(base);
        this.base = base;

        assertNotNull(messageCodec);
        this.messageCodec = messageCodec;

//        this.in = bufferingRequired ? new BufferedInputStream(new SyncRawConnection.ConnectionInputStream(this.base)) : new SyncRawConnection.ConnectionInputStream(this.base);
        this.in = new SyncRawConnection.ConnectionInputStream(this.base);
//        this.out = bufferingRequired ? new BufferedOutputStream(new SyncRawConnection.ConnectionOutputStream(this.base)) : new SyncRawConnection.ConnectionOutputStream(this.base);
        this.out = new SyncRawConnection.ConnectionOutputStream(this.base);

        this.closed = 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();
    }

    public M read()
            throws ConnectionException {
        try {
            return this.messageCodec.read(this.in);
        } catch (InterruptedIOException e) {
            throw new ReadTimeoutException(e);
        } catch (java.io.EOFException e) {
            throw new EOFException();
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    public void write(M message)
            throws ConnectionException {
        try {
            this.messageCodec.write(message, this.out);
            this.out.flush();
        } catch (Exception e) {
            throw new WriteException(e);
        }
    }

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

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

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

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

}
