/*
 * 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. 
 */

/*
 * @(#)HelloHandler.java	1.71 06/28/07
 */ 

package com.sun.messaging.jmq.jmsserver.data.handlers;

import java.util.*;
import java.net.*;
import com.sun.messaging.jmq.auth.api.FailedLoginException;
import com.sun.messaging.jmq.jmsserver.data.PacketHandler;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.util.GoodbyeReason;

import com.sun.messaging.jmq.io.*;
import com.sun.messaging.jmq.net.*;
import com.sun.messaging.jmq.util.ServiceType;
import com.sun.messaging.jmq.jmsserver.service.Connection;
import com.sun.messaging.jmq.jmsserver.service.ConnectionUID;
import com.sun.messaging.jmq.jmsserver.service.imq.IMQConnection;
import com.sun.messaging.jmq.jmsserver.service.imq.IMQBasicConnection;
import com.sun.messaging.jmq.jmsserver.service.imq.IMQService;
import com.sun.messaging.jmq.jmsserver.service.ConnectionManager;
import com.sun.messaging.jmq.jmsserver.service.HAMonitorService;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;
import com.sun.messaging.jmq.jmsserver.auth.AccessController;
import com.sun.messaging.jmq.jmsserver.auth.AuthCacheData;
import com.sun.messaging.jmq.jmsserver.util.memory.MemoryManager;
import com.sun.messaging.jmq.jmsserver.license.*;
import com.sun.messaging.jmq.jmsserver.cluster.*;
import com.sun.messaging.jmq.util.UID;
import com.sun.messaging.jmq.util.GoodbyeReason;




/**
 * Handler class which deals with the Hello message which is created when
 * a client starts talking to the broker.
 * Hello provides a "ping" message and also sets up a connection if the
 * protocol used does not have a unique way of determining a connection
 * (e.g. tcp does not need the HELLO message to set up a connection, since
 * each socket corresponds to a new connection)
 */
public class HelloHandler extends PacketHandler 
{
    private ConnectionManager connectionList;

    private Logger logger = Globals.getLogger();
    private BrokerResources rb = Globals.getBrokerResources();
    private static boolean DEBUG = false;

    private static boolean ALLOW_C_CLIENTS = false;
    private static boolean CAN_RECONNECT = false;

    static {
        try {
            LicenseBase license = Globals.getCurrentLicense(null);
            ALLOW_C_CLIENTS = license.getBooleanProperty(
                                license.PROP_ENABLE_C_API, false);
        } catch (BrokerException ex) {
            ALLOW_C_CLIENTS = false;
        }
        try {
            LicenseBase license = Globals.getCurrentLicense(null);
            CAN_RECONNECT = license.getBooleanProperty(
                                license.PROP_ENABLE_RECONNECT, false);
        } catch (BrokerException ex) {
            CAN_RECONNECT = false;
        }

    }

    public static void DUMP(String title) {
        Globals.getLogger().log(Logger.DEBUG,title);
        Globals.getLogger().log(Logger.DEBUG,"------------------------");
        Globals.getLogger().log(Logger.DEBUG,"Number of connections is " + 
                          Globals.getConnectionManager().getNumConnections(null));
        List l = Globals.getConnectionManager().getConnectionList(null);
        for (int i=0; i < l.size(); i ++ ) {
            Connection c= (Connection)l.get(i);
            Globals.getLogger().log(Logger.DEBUG,"\t" + i + "\t" +
                 c.getConnectionUID() + " :" + c.getRemoteConnectionString());
        }
        Globals.getLogger().log(Logger.DEBUG,"------------------------");
    }



    public HelloHandler(ConnectionManager list)
    {
        connectionList = list;
    }

    /**
     * Method to handle HELLO messages
     */
    public boolean handle(IMQConnection con, Packet msg) 
        throws BrokerException 
    { 

         if (DEBUG) {
             logger.log(Logger.DEBUGHIGH, "HelloHandler: handle() [ Received Hello Message]");
          }

          String reason = null;
          Hashtable hello_props = null;


          try {
              hello_props = msg.getProperties();
          } catch (Exception ex) {
              
              logger.log(Logger.INFO,"Internal Error: error "
                   + " retrieving properties from hello message ", ex);
             hello_props = new Hashtable();

          }

          boolean alreadyStarted = con.isStarted();
          boolean alreadyAuthenticated = con.isAuthenticated();

          int requestedProtocol = 0;
          int highestProtocol = con.getHighestSupportedProtocol();
          int lowestProtocol = PacketType.VERSION1;

          String expectedClusterID = null;
          UID expectedSessionID = null;
          ConnectionUID oldCID = null;
          Integer bufsize = null;
          boolean badClientType = false;
          if (hello_props != null) {
              Integer level = (Integer)hello_props.get("JMQProtocolLevel");
              if (level == null) {
                  requestedProtocol=PacketType.VERSION1;
              } else {
                  requestedProtocol=level.intValue();
              }
               bufsize = (Integer)hello_props.get("JMQSize");
              if (bufsize == null) { //XXX try old protocol
                  bufsize = (Integer)hello_props.get("JMQRBufferSize");
              }

              // Retrieve HA related properties
              Long longUID = (Long)hello_props.get("JMQStoreSession");
              if (longUID != null) {
                  expectedSessionID = new UID(longUID.longValue());
              }

              expectedClusterID = (String)hello_props.get("JMQClusterID");

              Boolean reconnectable = (Boolean)hello_props.get("JMQReconnectable");
              Boolean haclient = (Boolean)hello_props.get("JMQHAClient");
              if (Globals.getHAEnabled() && haclient != null && haclient.booleanValue()) {
                  reconnectable = haclient;
              }

              String s = (String)hello_props.get("JMQUserAgent");
              if (!ALLOW_C_CLIENTS && s != null &&  s.indexOf("C;") != -1) {
                  badClientType = true;
              }
              if (s != null) {
                con.addClientData(IMQConnection.USER_AGENT, s);
              }

              longUID = (Long)hello_props.get("JMQConnectionID");

              if (longUID != null) {
                  logger.log(Logger.DEBUG,"Have old connectionUID");
                  oldCID = new ConnectionUID(longUID.longValue());
                  logger.log(Logger.INFO,
                         BrokerResources.I_RECONNECTING, oldCID);
                  logger.log(Logger.DEBUG,"Checking for active connection");

                  Connection oldcon = Globals.getConnectionManager().getConnection(oldCID);
                  DUMP("Before connection Destroy");
                  if (oldcon != null) {
                      logger.log(Logger.DEBUG,"Destroying old connection " + oldCID);
                      oldcon.destroyConnection(true,GoodbyeReason.ADMIN_KILLED_CON, "Destroying old connection with same connectionUID " + oldCID + " - reconnect is happening before connection was reaped");
                  }
/* LKS
                  DUMP();

                  logger.log(Logger.DEBUG,"Updating connection in id list " +
                           "["+oldcid + "," + uid + "]");
                  // old code
                  con.setConnectionUID(oldcid);
                  Globals.getConnectionManager().updateConnectionUID(
                         oldcid, uid);
                  //Globals.getConnectionManager().updateConnectionUID(
                  //       uid, oldcid);
*/
                  DUMP("After Connection Destroy");
              }

              con.getConnectionUID().setCanReconnect(reconnectable == null ? false :
                      reconnectable.booleanValue());

              Long interval = (Long)hello_props.get("JMQInterval");

              // LKS - XXX just override for testing
              long itime = interval == null ? 10000 : interval.intValue();
              con.setReconnectInterval(itime);
               
          } else {
              requestedProtocol=PacketType.VERSION1;
          }

          int supportedProtocol = 0;
          if (requestedProtocol > highestProtocol) {
              supportedProtocol = highestProtocol;
          } else if (requestedProtocol < lowestProtocol) {
              supportedProtocol = lowestProtocol;
          }  else {
              supportedProtocol = requestedProtocol;
          }
          con.setClientProtocolVersion(supportedProtocol);

           if (bufsize != null) {
              logger.log(Logger.DEBUG, "Received JMQRBufferSize -" + bufsize);
              con.setFlowCount(bufsize.intValue());
           }


          Packet pkt = new Packet(con.useDirectBuffers());
          pkt.setPacketType(PacketType.HELLO_REPLY);
          pkt.setConsumerID(msg.getConsumerID());
          Hashtable hash = new Hashtable();
          reason = "unavailable";
          int status = Status.UNAVAILABLE;

          // If the connection's remote IP address was not set by the
          // protocol, then use the IP in the message packet.
          if (con.getRemoteIP() == null) {
            con.setRemoteIP(msg.getIP());
          }


          if ((alreadyAuthenticated || alreadyStarted) 
                  && !msg.getIndempotent() ) { // handle ibit
              status = Status.ERROR;
              reason = "Connection reuse not allowed";
              if (alreadyAuthenticated) {
                  logger.log(Logger.WARNING,"Internal Error: " +
                    " received HELLO on already authenticated connection "
                    + con.getRemoteConnectionString() +
                    " " + con.getConnectionUID());
              } else {
                  logger.log(Logger.WARNING,"Internal Error: " +
                    " received HELLO on already started connection "
                    + con.getRemoteConnectionString() +
                    " " + con.getConnectionUID());
              }
          
          } else if (badClientType) {
              logger.log(Logger.ERROR, rb.E_FEATURE_UNAVAILABLE,
                   Globals.getBrokerResources().getString(
                        BrokerResources.M_C_API));
              reason = "C clients not allowed on this version";
	      status = Status.UNAVAILABLE;
          } else if (!CAN_RECONNECT && con.getConnectionUID().getCanReconnect()) {
              logger.log(Logger.ERROR, rb.E_FEATURE_UNAVAILABLE,
                   Globals.getBrokerResources().getString(
                        BrokerResources.M_CLIENT_FAILOVER));
              reason = "Client Failover not allowed on this version";
          } else if (requestedProtocol != supportedProtocol) {
              // Bad protocol level.
              logger.log(Logger.WARNING, rb.W_BAD_PROTO_VERSION,
                Integer.toString(requestedProtocol), 
                Integer.toString(supportedProtocol));

              reason = "bad version";
	      status = Status.BAD_VERSION;
          } else if (con.getConnectionState() != Connection.STATE_UNAVAILABLE) {
              /** 
               * connection may not be able to be created e.g: 
               * licensing, being destroyed (e.g due to timeout)
               */
              if (con.setConnectionState(Connection.STATE_INITIALIZED)) {
                  reason = null;
                  status = Status.OK;
              } else {
                  status = Status.UNAVAILABLE;
	      }
          } else {
	      status = Status.UNAVAILABLE;
          }
          
          UID brokerSessionID = Globals.getBrokerSessionID();
          if (brokerSessionID!=null){
        	  hash.put("JMQBrokerSessionID",new Long(brokerSessionID.longValue()));
          }
          
          // OK, handle the HA properties HERE

          String clusterID = null;
          UID sessionUID = null;
          Set supportedSessions = null;

          ClusterManager cfg = Globals.getClusterManager();
          if (cfg != null) {
              clusterID = cfg.getClusterId();
              sessionUID = cfg.getStoreSessionUID();

              hash.put("JMQHA",Boolean.valueOf(cfg.isHA()));

              if (clusterID != null) {
                  hash.put("JMQClusterID", clusterID);
              }
              if (sessionUID != null)  {
                  hash.put("JMQStoreSession",new Long(sessionUID.longValue()));
              }

              String list = null;
              Iterator itr = null;
              if (((IMQService)con.getService()).getServiceType() != ServiceType.ADMIN) {
                  itr = cfg.getKnownBrokers(false);
              } else {
                  itr = cfg.getKnownBrokers(true);
              }
              Set s = new HashSet();
              // ok get rid of dups
              while (itr.hasNext()) {
                  ClusteredBroker cb = (ClusteredBroker)itr.next();
                  s.add(cb.getBrokerURL().toString());
              }
              // OK .. now convert to a string

              itr = s.iterator();

              while (itr.hasNext()) {
                  if (list == null) {
                      list = itr.next().toString();
                  } else {
                      list += "," + itr.next().toString();
                  }
              }
              if (list != null) {
                  hash.put("JMQBrokerList", list);
              }   
          }

          HAMonitorService hamonitor = Globals.getHAMonitorService(); 
          if (hamonitor != null && hamonitor.inTakeover()) {
              if (((IMQService)con.getService()).getServiceType() != ServiceType.ADMIN) {
                  status = Status.TIMEOUT;
                  if (oldCID != null) {
                      logger.log(logger.INFO,
                          BrokerResources.W_IN_TAKEOVER_RECONNECT_LATER, oldCID);
                  } else {
                      logger.log(logger.INFO,
                          BrokerResources.W_IN_TAKEOVER_RECONNECT_LATER,
                          con.getConnectionUID());
                  }
              }
          }
          
          // first we want to deal with a bad clusterid
          if (clusterID != null && expectedClusterID != null
                 && !clusterID.equals(expectedClusterID)) {
              status = Status.BAD_REQUEST;

          } else if (expectedSessionID != null && sessionUID != null &&
                     expectedSessionID.equals(sessionUID)) {
               // cool we connected to the right broker
               // we already have the right owner
          } else if (expectedSessionID != null) {


              if (cfg == null) { // not running any cluster config
                      logger.log(Logger.WARNING, 
                            BrokerResources.E_INTERNAL_BROKER_ERROR, 
                              "Internal Error: Received session on"
                            + " non-clustered broker");

                      status = Status.NOT_FOUND;

              } else {
                  // OK, if we are here, we need to locate the right
                  // broker for the session
                  //
                  // Here are the steps we need to check:
                  //  1. does this broker support the sessionUID
                  //     if not
                  //  2. can we locate another broker with the sessionUID
                  //

                  ClusteredBroker owner = null;

                  //
                  // OK, see if this was a session UID we took over at some
                  // point in the past
                  Set s = cfg.getSupportedStoreSessionUIDs();
                  if (s.contains(expectedSessionID)) {
                      // yep, we took it over
                      owner = cfg.getLocalBroker();
                  }

                  // OK, we dont have it in our list
                  //
                  // We want to find out who did take it over (although that
                  // information may be lost, there are no guarentees at this
                  // point since there is a limit to what we persistently store
    
                  if (owner == null) { // this broker isnt supprting the session
    
                      // see if the database indicates someone else has it
    
                      String ownerString = cfg.lookupStoreSessionOwner(expectedSessionID);
                      if (ownerString != null) {
                          owner = cfg.getBroker(ownerString);
                      }
                  }
                  try {
    
                      // ok, we didnt find it (that doesnt mean someone didnt
                      // take it over, just that we cant find out who)
                      // makesure we return the right error message
     
                      if (owner != null) {
                          ClusteredBroker creator = null;
                          String creatorString = cfg.getStoreSessionCreator(expectedSessionID); 
                          if (creatorString != null) {
                              creator = cfg.getBroker(creatorString);
                          }
                          int stat = owner.getStatus();
                          if (BrokerStatus.getBrokerInDoubt(stat) ||
                              !BrokerStatus.getBrokerLinkIsUp(stat) ||
                              owner.getState() == BrokerState.FAILOVER_STARTED) {
                              status = Status.TIMEOUT;
                              logger.log(logger.INFO, Globals.getBrokerResources().getKString(
                              BrokerResources.I_RECONNECT_OWNER_INDOUBT, expectedSessionID, owner));
                          } else if (!owner.isLocalBroker()) {
                              status = Status.MOVED_PERMANENTLY;
                              hash.put("JMQStoreOwner", owner.getBrokerURL().toString());
                              logger.log(logger.INFO, Globals.getBrokerResources().getKString(
                                     BrokerResources.I_RECONNECT_OWNER_NOTME, expectedSessionID, owner));
                          } else if (creator == null) { //XXX
                              status = Status.NOT_FOUND;
                              logger.log(logger.INFO, Globals.getBrokerResources().getKString(
                                         BrokerResources.I_RECONNECT_NOCREATOR, expectedSessionID));
                          } else if (creator.getState() == BrokerState.FAILOVER_STARTED) {
                              status = Status.TIMEOUT;
                              logger.log(logger.INFO, Globals.getBrokerResources().getKString(
                                         BrokerResources.I_RECONNECT_INTAKEOVER, expectedSessionID));
                          } else { // local broker owns us - set owner for debugging only
                                   // not required for protocol
                              hash.put("JMQStoreOwner", owner.getBrokerURL().toString());
                          }
                      } else { // didnt find owner
                          status = Status.NOT_FOUND;
                          logger.log(logger.INFO, Globals.getBrokerResources().getKString(
                                  BrokerResources.I_RECONNECT_OWNER_NOTFOUND, expectedSessionID));
                      }
                  } catch (Exception ex) {
                      logger.log(Logger.WARNING, 
                            BrokerResources.W_RECONNECT_ERROR, expectedSessionID.toString(), ex);
                      status = Status.NOT_FOUND;
                  }
             }

          }


          if (!con.isAdminConnection() && Globals.getMemManager() != null) {

              hash.put("JMQSize", new Integer(Globals.getMemManager().getJMQSize()));
              hash.put("JMQBytes", new Long(Globals.getMemManager().getJMQBytes()));
              hash.put("JMQMaxMsgBytes", new Long(Globals.getMemManager().getJMQMaxMsgBytes()));
          }
          if (reason != null) {
              hash.put("JMQReason", reason);
          }
          hash.put("JMQService", con.getService().getName());
          hash.put("JMQStatus", new Integer(status));
          hash.put("JMQConnectionID", new Long(con.getConnectionUID().longValue()));
          hash.put("JMQProtocolLevel",
                       new Integer(supportedProtocol));
          hash.put("JMQVersion",
                       Globals.getVersion().getProductVersion());
          if (((IMQBasicConnection)con).getDumpPacket() ||
                 ((IMQBasicConnection)con).getDumpOutPacket())
              hash.put("JMQReqID", msg.getSysMessageID().toString());

          try {
              // Added licensing description properties
              LicenseBase license = Globals.getCurrentLicense(null);
              hash.put("JMQLicense",
                  license.getProperty(LicenseBase.PROP_LICENSE_TYPE));
              hash.put("JMQLicenseDesc",
                  license.getProperty(LicenseBase.PROP_DESCRIPTION));
          } catch (BrokerException ex) {
              // This should never happen, but go ahead and at least
              // capture exception here
              hash.put("JMQLicenseDesc", ex.toString());
          }

          pkt.setProperties(hash);

	  con.sendControlMessage(pkt);

         // OK .. valid status messages are
         if (status != Status.OK && status != Status.MOVED_PERMANENTLY
              && status != Status.NOT_FOUND && status != Status.TIMEOUT) {
             // destroy the connection !!! (should be ok if destroy twice)
             con.closeConnection(true, GoodbyeReason.CON_FATAL_ERROR,
                  Globals.getBrokerResources().getKString(
                  BrokerResources.M_INIT_FAIL_CLOSE));
             connectionList.removeConnection(con.getConnectionUID(),
                  false,GoodbyeReason.CON_FATAL_ERROR,
                  Globals.getBrokerResources().getKString(
                  BrokerResources.M_INIT_FAIL_CLOSE));
             return true;
         }

          status = Status.UNAVAILABLE;
          String authType = null;
          if (hello_props != null) {
              authType = (String)hello_props.get("JMQAuthType");
          }
          AccessController ac = con.getAccessController();
          pkt = new Packet(con.useDirectBuffers());
          pkt.setPacketType(PacketType.AUTHENTICATE_REQUEST);
          pkt.setConsumerID(msg.getConsumerID());

          hash = new Hashtable();
          hash.put("JMQSequence", new Integer(msg.getSequence()));
          hash.put("JMQChallenge", Boolean.valueOf(true));

          Properties props = new Properties();
          props.setProperty(Globals.IMQ + ".clientIP", msg.getIPString());
          props.setProperty(Globals.IMQ + ".connectionID", con.getConnectionUID().toString());
          byte[] req = null;
          try {
          AuthCacheData acd = ((IMQService)con.getService()).getAuthCacheData();
          req = ac.getChallenge(msg.getSequence(), props, 
                                acd.getCacheData(), authType);
          hash.put("JMQAuthType", ac.getAuthType());
          if (con.setConnectionState(Connection.STATE_AUTH_REQUESTED)) {
          status = Status.OK;
          }

          } catch (FailedLoginException e) {
          logger.log(Logger.WARNING, e.getMessage(), e);
          status = Status.FORBIDDEN;
          } catch (OutOfMemoryError err) {
              // throw error so that memory is freed and
              // packet is re-processed
              throw err;
          } catch (Throwable w) {
          logger.log(Logger.ERROR, Globals.getBrokerResources().getKString(
              BrokerResources.E_GET_CHALLENGE_FAILED)+" - "+w.getMessage(), w);
          status = Status.FORBIDDEN;
          }
          hash.put("JMQStatus", new Integer(status));
          if (((IMQBasicConnection)con).getDumpPacket() ||
                ((IMQBasicConnection)con).getDumpOutPacket()) 
              hash.put("JMQReqID", msg.getSysMessageID().toString());

          pkt.setProperties(hash);
          if (req != null) {
              pkt.setMessageBody(req);
          }
          
          con.sendControlMessage(pkt);
          if (DEBUG) {
          logger.log(Logger.DEBUG,  "HelloHandler: handle() [ sent challenge ]"
                                      + ":status="+ Status.getString(status));
          }

          if (status != Status.OK && status != Status.MOVED_PERMANENTLY
              && status != Status.NOT_FOUND && status != Status.TIMEOUT) {
             // destroy the connection !!! (should be ok if destroy twice)
              con.closeConnection(true, GoodbyeReason.CON_FATAL_ERROR, 
                  Globals.getBrokerResources().getKString(
                  BrokerResources.M_INIT_FAIL_CLOSE));
              connectionList.removeConnection(con.getConnectionUID(),
                  false,GoodbyeReason.CON_FATAL_ERROR,
                  Globals.getBrokerResources().getKString(
                  BrokerResources.M_INIT_FAIL_CLOSE));
          }
          return true;

    }

}
