/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2000-2009 Sun Microsystems, Inc. All rights reserved. 
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License ("CDDL") (collectively, the "License").  You may
 * not use this file except in compliance with the License.  You can obtain
 * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
 * or mq/legal/LICENSE.txt.  See the License for the specific language
 * governing permissions and limitations under the License.
 * 
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at mq/legal/LICENSE.txt.  Sun designates
 * this particular file as subject to the "Classpath" exception as provided by
 * Sun in the GPL Version 2 section of the License file that accompanied this
 * code.  If applicable, add the following below the License Header, with the
 * fields enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 * 
 * Contributor(s):
 * 
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or  to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright holder. 
 */

/*
 * @(#)ReadThread.java	1.14 06/29/07
 */ 

package com.sun.messaging.jmq.jmsserver.service.imq.group;

import java.util.*;
import java.io.*;
import java.nio.channels.spi.*;
import java.nio.channels.*;
import com.sun.messaging.jmq.util.log.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.service.imq.*;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.service.*;
import com.sun.messaging.jmq.jmsserver.util.pool.*;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;


class ReadThread extends SelectThread
{
    int selector_cnt = 0;

    public ReadThread(Service svc, MapEntry entry) 
        throws IOException
    {
        super(svc, entry);

        type = "read";
        INITIAL_KEY=SelectionKey.OP_READ; // none
        POSSIBLE_MASK=SelectionKey.OP_READ; // none
    } 

    public Hashtable getDebugState() {
        Hashtable ht = new Hashtable();
        ht.put("TYPE", "ReadThread");
        ht.put("selector_cnt", new Integer(selector_cnt));
        return ht;
    }


    protected void wakeup() {
        Selector s = selector;
        if (s != null)
            s.wakeup();
    }

    protected void process() 
        throws IOException
    {
       Selector s = selector;
       if (s == null)
          throw new IOException("connection gone");
       int cnt =  0;
       try {
          cnt = s.select(TIMEOUT);
       } catch (java.nio.channels.CancelledKeyException ex) {
         // bug 4944894
         // nio can throw the cancelledKeyException all the
         // way up in some cases, this does not indicate that
         // the selector is closed so the broker should ignore
         // the issue

          return;
       }
       if (cnt > 0) {
           Set keys = s.selectedKeys();
           Iterator keyitr = keys.iterator();
           while (keyitr.hasNext()) {
               SelectionKey key = (SelectionKey)keyitr.next();
               IMQIPConnection con = (IMQIPConnection)key.attachment();
               try {
                   int result =  con.readData();
                   // triggers bug 4708106
                   //if (result == Operation.PROCESS_WRITE_INCOMPLETE) {
                       keyitr.remove();
                   //}
               } catch (BrokerException ex) {
                   removeConnection(con, ex.getMessage());
                   keyitr.remove();
               } catch (IOException ex) {
                   String reason = (con.getDestroyReason() == null ?
                        (ex instanceof EOFException ? 
                            Globals.getBrokerResources().getKString(
                    BrokerResources.M_CONNECTION_CLOSE) 
                       : ex.toString()) : con.getDestroyReason());
                   removeConnection(con, reason);
                   keyitr.remove();
               }
           }
       }
    }

}
