001/*
002 *  Licensed to the Apache Software Foundation (ASF) under one
003 *  or more contributor license agreements.  See the NOTICE file
004 *  distributed with this work for additional information
005 *  regarding copyright ownership.  The ASF licenses this file
006 *  to you under the Apache License, Version 2.0 (the
007 *  "License"); you may not use this file except in compliance
008 *  with the License.  You may obtain a copy of the License at
009 *
010 *    http://www.apache.org/licenses/LICENSE-2.0
011 *
012 *  Unless required by applicable law or agreed to in writing,
013 *  software distributed under the License is distributed on an
014 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 *  KIND, either express or implied.  See the License for the
016 *  specific language governing permissions and limitations
017 *  under the License.
018 *
019 */
020package org.apache.mina.filter.stream;
021
022import java.io.IOException;
023import java.util.Queue;
024import java.util.concurrent.ConcurrentLinkedQueue;
025
026import org.apache.mina.core.buffer.IoBuffer;
027import org.apache.mina.core.filterchain.IoFilterAdapter;
028import org.apache.mina.core.filterchain.IoFilterChain;
029import org.apache.mina.core.session.AttributeKey;
030import org.apache.mina.core.session.IoSession;
031import org.apache.mina.core.write.DefaultWriteRequest;
032import org.apache.mina.core.write.WriteRequest;
033
034/**
035 * TODO Add documentation
036 * 
037 * @author <a href="http://mina.apache.org">Apache MINA Project</a>
038 */
039public abstract class AbstractStreamWriteFilter<T> extends IoFilterAdapter {
040    /**
041     * The default buffer size this filter uses for writing.
042     */
043    public static final int DEFAULT_STREAM_BUFFER_SIZE = 4096;
044
045    /**
046     * The attribute name used when binding the streaming object to the session.
047     */
048    protected final AttributeKey CURRENT_STREAM = new AttributeKey(getClass(), "stream");
049
050    protected final AttributeKey WRITE_REQUEST_QUEUE = new AttributeKey(getClass(), "queue");
051
052    protected final AttributeKey CURRENT_WRITE_REQUEST = new AttributeKey(getClass(), "writeRequest");
053
054    private int writeBufferSize = DEFAULT_STREAM_BUFFER_SIZE;
055
056    @Override
057    public void onPreAdd(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
058        Class<? extends IoFilterAdapter> clazz = getClass();
059        if (parent.contains(clazz)) {
060            throw new IllegalStateException("Only one " + clazz.getName() + " is permitted.");
061        }
062    }
063
064    @Override
065    public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
066        // If we're already processing a stream we need to queue the WriteRequest.
067        if (session.getAttribute(CURRENT_STREAM) != null) {
068            Queue<WriteRequest> queue = getWriteRequestQueue(session);
069            if (queue == null) {
070                queue = new ConcurrentLinkedQueue<WriteRequest>();
071                session.setAttribute(WRITE_REQUEST_QUEUE, queue);
072            }
073            queue.add(writeRequest);
074            return;
075        }
076
077        Object message = writeRequest.getMessage();
078
079        if (getMessageClass().isInstance(message)) {
080
081            T stream = getMessageClass().cast(message);
082
083            IoBuffer buffer = getNextBuffer(stream);
084            if (buffer == null) {
085                // End of stream reached.
086                writeRequest.getFuture().setWritten();
087                nextFilter.messageSent(session, writeRequest);
088            } else {
089                session.setAttribute(CURRENT_STREAM, message);
090                session.setAttribute(CURRENT_WRITE_REQUEST, writeRequest);
091
092                nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
093            }
094
095        } else {
096            nextFilter.filterWrite(session, writeRequest);
097        }
098    }
099
100    abstract protected Class<T> getMessageClass();
101
102    @SuppressWarnings("unchecked")
103    private Queue<WriteRequest> getWriteRequestQueue(IoSession session) {
104        return (Queue<WriteRequest>) session.getAttribute(WRITE_REQUEST_QUEUE);
105    }
106
107    @SuppressWarnings("unchecked")
108    private Queue<WriteRequest> removeWriteRequestQueue(IoSession session) {
109        return (Queue<WriteRequest>) session.removeAttribute(WRITE_REQUEST_QUEUE);
110    }
111
112    @Override
113    public void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
114        T stream = getMessageClass().cast(session.getAttribute(CURRENT_STREAM));
115
116        if (stream == null) {
117            nextFilter.messageSent(session, writeRequest);
118        } else {
119            IoBuffer buffer = getNextBuffer(stream);
120
121            if (buffer == null) {
122                // End of stream reached.
123                session.removeAttribute(CURRENT_STREAM);
124                WriteRequest currentWriteRequest = (WriteRequest) session.removeAttribute(CURRENT_WRITE_REQUEST);
125
126                // Write queued WriteRequests.
127                Queue<WriteRequest> queue = removeWriteRequestQueue(session);
128                if (queue != null) {
129                    WriteRequest wr = queue.poll();
130                    while (wr != null) {
131                        filterWrite(nextFilter, session, wr);
132                        wr = queue.poll();
133                    }
134                }
135
136                currentWriteRequest.getFuture().setWritten();
137                nextFilter.messageSent(session, currentWriteRequest);
138            } else {
139                nextFilter.filterWrite(session, new DefaultWriteRequest(buffer));
140            }
141        }
142    }
143
144    /**
145     * @return the size of the write buffer in bytes. Data will be read from the
146     * stream in chunks of this size and then written to the next filter.
147     */
148    public int getWriteBufferSize() {
149        return writeBufferSize;
150    }
151
152    /**
153     * Sets the size of the write buffer in bytes. Data will be read from the
154     * stream in chunks of this size and then written to the next filter.
155     *
156     * @param writeBufferSize The size of the write buffer
157     * @throws IllegalArgumentException if the specified size is &lt; 1.
158     */
159    public void setWriteBufferSize(int writeBufferSize) {
160        if (writeBufferSize < 1) {
161            throw new IllegalArgumentException("writeBufferSize must be at least 1");
162        }
163        this.writeBufferSize = writeBufferSize;
164    }
165
166    abstract protected IoBuffer getNextBuffer(T message) throws IOException;
167}