pipesession.java

银行项目为后台ｓｏｃｋｅｔ通信写的程序

/****************************************************************************
 * Package		: com.ecSolutions.ecAppServer.server.session.nio
 * File			: PipeSession.java
 * Create Date  : 2007-7-20
 * Author		: Steven Chen
 * 
 * Copyright(C) 2006 ecSolutions(shanghai) Co.,Limited.All Rights Reserved.
 *			
 ***************************************************************************/
package com.ecSolutions.ecAppServer.server.session.nio;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.Pipe;
import java.nio.channels.SelectableChannel;
import java.nio.channels.Pipe.SinkChannel;
import java.nio.channels.Pipe.SourceChannel;

import com.ecSolutions.ecAppServer.server.Buffer;
import com.ecSolutions.ecAppServer.server.Packet;
import com.ecSolutions.ecAppServer.server.SessionType;
import com.ecSolutions.ecAppServer.server.buffer.BufferFactory;
import com.ecSolutions.ecAppServer.server.packet.DefaultPacket;
import com.ecSolutions.ecAppServer.server.session.nio.reactor.ReactorHandler;
import com.ecSolutions.ecAppServer.server.util.ChannelUtils;

/**
 * Pipe session.
 * 
 * @author Steven Chen
 * @version $Id: PipeSession.java,v 1.2 2007/07/26 03:48:59 stevenchen Exp $
 */
public class PipeSession extends AbstractChannelSession {

    private Pipe pipe;

    /**
     * Set the pipe which the session will used.
     * 
     * @param pipe
     *            pipe
     * @throws IllegalStateException
     */
    public void setPipe(Pipe pipe) {
        if (isStarted())
            throw new IllegalStateException(
                    "can't set pipe after session started");
        this.pipe = pipe;
    }

    /**
     * Get pipe associted with the session.
     * 
     * @return pipe
     */
    public Pipe getPipe() {
        return pipe;
    }

    public SessionType getSessionType() {
        return SessionType.PIPE;
    }

    public SocketAddress getLocalAddress() {
        return null;
    }

    public SocketAddress getRemoteAddress() {
        return null;
    }

    protected ReactorHandler getReactorHandler() {
        return new ChannelReactorHandler() {

            public SelectableChannel[] getChannels() {
                return new SelectableChannel[] { pipe.sink(), pipe.source() };
            }

            protected void read() throws IOException {
                Buffer buffer = BufferFactory.allocate(getReadPacketSize());
                SourceChannel source = pipe.source();
                int n = -1;
                int readCount = 0;

                try {
                    while ((n = buffer.read(source)) >= 0) {
                        if (n == 0)
                            break;
                        readCount += n;
                    }
                } catch (IOException e) {
                    buffer.release();
                    throw e;
                }

                if (readCount > 0) {
                    buffer.flip();
                    getSessionFilterChain(false).packetReceived(
                            new DefaultPacket(buffer));
                }
                if (n < 0) // Connection closed
                    throw new ClosedChannelException();
            }

            protected boolean write(Packet packet) throws IOException {
                Buffer buffer = packet.getContent();
                SinkChannel sink = pipe.sink();
                while (true) {
                    int n = buffer.write(sink);
                    if (!buffer.hasRemaining())
                        return true;
                    else if (n == 0) {
                        // have more data, but the kennel buffer
                        // is full, wait next time to write
                        return false;
                    }
                }
            }

        };
    }

    protected void doStart() throws IOException {
        if (pipe == null) {
            try {
                pipe = Pipe.open();
            } catch (IOException e) {
                doClose();
                throw e;
            }
        }
    }

    protected void doClose() {
        if (pipe != null) {
            ChannelUtils.close(pipe.sink());
            ChannelUtils.close(pipe.source());
            pipe = null;
        }
    }
}