/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/

package org.apache.james.protocols.impl;

import java.io.InputStream;
import java.net.InetSocketAddress;

import javax.net.ssl.SSLEngine;

import org.apache.james.protocols.api.AbstractProtocolTransport;
import org.apache.james.protocols.api.ProtocolSession;
import org.apache.james.protocols.api.handler.LineHandler;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.handler.ssl.SslHandler;
import org.jboss.netty.handler.stream.ChunkedStream;

/**
 * A Netty implementation of a ProtocolTransport
 */
public class NettyProtocolTransport extends AbstractProtocolTransport {
    
    private final Channel channel;
    private final SSLEngine engine;
    private int lineHandlerCount = 0;
    
    public NettyProtocolTransport(Channel channel, SSLEngine engine) {
        this.channel = channel;
        this.engine = engine;
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#getRemoteAddress()
     */
    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) channel.getRemoteAddress();
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#getId()
     */
    public String getId() {
        return channel.getId() + "";
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#isTLSStarted()
     */
    public boolean isTLSStarted() {
        if (isStartTLSSupported()) {
            return channel.getPipeline().get("sslHandler") != null;
        } 
        return false;
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#isStartTLSSupported()
     */
    public boolean isStartTLSSupported() {
        return engine != null;
    }


    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#popLineHandler()
     */
    public void popLineHandler() {
        if (lineHandlerCount > 0) {
            channel.getPipeline().remove("lineHandler" + lineHandlerCount);
            lineHandlerCount--;
        }
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#pushLineHandler(org.apache.james.protocols.api.handler.LineHandler, org.apache.james.protocols.api.ProtocolSession)
     */
    public <T extends ProtocolSession> void pushLineHandler(LineHandler<T> overrideCommandHandler,
            T session) {
        lineHandlerCount++;
        // Add the linehandler in front of the coreHandler so we can be sure 
        // it is executed with the same ExecutorHandler as the coreHandler (if one exist)
        // 
        // See JAMES-1277
        channel.getPipeline().addBefore("coreHandler", "lineHandler" + lineHandlerCount, new LineHandlerUpstreamHandler<T>(session, overrideCommandHandler));
    }

    /**
     * @see org.apache.james.protocols.api.ProtocolTransport#getPushedLineHandlerCount()
     */
    public int getPushedLineHandlerCount() {
        return lineHandlerCount;
    }

    

    @Override
    protected void startTLS(ProtocolSession session) {
        channel.setReadable(false);
        SslHandler filter = new SslHandler(engine);
        filter.getEngine().setUseClientMode(false);
        channel.getPipeline().addFirst("sslHandler", filter);
        channel.setReadable(true);        
    }

    @Override
    protected void writeToClient(byte[] bytes, ProtocolSession session) {
        channel.write(ChannelBuffers.wrappedBuffer(bytes));
    }

    @Override
    protected void close() {
        channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }


    @Override
    protected void writeToClient(InputStream in, ProtocolSession session) {
        channel.write(new ChunkedStream(in));
    }

	@Override
	public void setReadable(boolean readable) {
		channel.setReadable(readable);
	}

	@Override
	public boolean isReadable() {
		return channel.isReadable();
	}

	@Override
	public InetSocketAddress getLocalAddress() {
		return (InetSocketAddress) channel.getLocalAddress();
	}
    
    
    
}

@@ -5056,25 +5056,28 @@
 ;%0D%0A    %7D%0D%0A%0D%0A
-%09
+    
 @Override%0D%0A%09
@@ -5071,25 +5071,28 @@
  @Override%0D%0A
-%09
+    
 public void 
@@ -5124,18 +5124,24 @@
 ble) %7B%0D%0A
-%09%09
+        
 channel.
@@ -5160,31 +5160,37 @@
 readable);%0D%0A
-%09
+    
 %7D%0D%0A%0D%0A
-%09
+    
 @Override%0D%0A%09
@@ -5184,25 +5184,28 @@
  @Override%0D%0A
-%09
+    
 public boole
@@ -5219,26 +5219,32 @@
 adable() %7B%0D%0A
-%09%09
+        
 return chann
@@ -5265,15 +5265,21 @@
 );%0D%0A
-%09
+    
 %7D%0D%0A%0D%0A
-%09
+    
 @Ove
@@ -5285,17 +5285,20 @@
 erride%0D%0A
-%09
+    
 public I
@@ -5339,10 +5339,16 @@
  %7B%0D%0A
-%09%09
+        
 retu
@@ -5402,9 +5402,12 @@
 );%0D%0A
-%09
+    
 %7D%0D%0A 

Replace tabs with spaces