geode-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java - geode - Git at Google

 /*
  * 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.geode.redis.internal.netty;


 import java.io.IOException;
 import java.math.BigInteger;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Supplier;

 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.EventLoopGroup;
 import io.netty.handler.codec.DecoderException;
 import org.apache.logging.log4j.Logger;

 import org.apache.geode.ForcedDisconnectException;
 import org.apache.geode.cache.CacheClosedException;
 import org.apache.geode.cache.execute.FunctionException;
 import org.apache.geode.cache.execute.FunctionInvocationTargetException;
 import org.apache.geode.distributed.DistributedSystemDisconnectedException;
 import org.apache.geode.logging.internal.log4j.api.LogService;
 import org.apache.geode.redis.internal.GeodeRedisServer;
 import org.apache.geode.redis.internal.ParameterRequirements.RedisParametersMismatchException;
 import org.apache.geode.redis.internal.RedisCommandType;
 import org.apache.geode.redis.internal.RedisConstants;
 import org.apache.geode.redis.internal.RegionProvider;
 import org.apache.geode.redis.internal.data.RedisDataTypeMismatchException;
 import org.apache.geode.redis.internal.executor.CommandFunction;
 import org.apache.geode.redis.internal.executor.RedisResponse;
 import org.apache.geode.redis.internal.pubsub.PubSub;
 import org.apache.geode.redis.internal.statistics.RedisStats;

 /**
  * This class extends {@link ChannelInboundHandlerAdapter} from Netty and it is the last part of the
  * channel pipeline. The {@link ByteToCommandDecoder} forwards a {@link Command} to this class which
  * executes it and sends the result back to the client. Additionally, all exception handling is done
  * by this class.
  * <p>
  * Besides being part of Netty's pipeline, this class also serves as a context to the execution of a
  * command. It provides access to the {@link RegionProvider} and anything else an executing {@link
  * Command} may need.
  */
 public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {

   private static final Logger logger = LogService.getLogger();
   private static final Command TERMINATE_COMMAND = new Command();

   private final Client client;
   private final Channel channel;
   private final RegionProvider regionProvider;
   private final PubSub pubsub;
   private final ByteBufAllocator byteBufAllocator;
   private final byte[] authPassword;
   private final Supplier<Boolean> allowUnsupportedSupplier;
   private final Runnable shutdownInvoker;
   private final RedisStats redisStats;
   private final EventLoopGroup subscriberGroup;
   private BigInteger scanCursor;
   private BigInteger sscanCursor;
   private BigInteger hscanCursor;
   private final AtomicBoolean channelInactive = new AtomicBoolean();
   private final int MAX_QUEUED_COMMANDS =
       Integer.getInteger("geode.redis.commandQueueSize", 1000);
   private final LinkedBlockingQueue<Command> commandQueue =
       new LinkedBlockingQueue<>(MAX_QUEUED_COMMANDS);

   private final int serverPort;
   private CountDownLatch eventLoopSwitched;

   private boolean isAuthenticated;

   /**
    * Default constructor for execution contexts.
    *
    * @param channel Channel used by this context, should be one to one
    * @param password Authentication password for each context, can be null
    */
   public ExecutionHandlerContext(Channel channel,
       RegionProvider regionProvider,
       PubSub pubsub,
       Supplier<Boolean> allowUnsupportedSupplier,
       Runnable shutdownInvoker,
       RedisStats redisStats,
       ExecutorService backgroundExecutor,
       EventLoopGroup subscriberGroup,
       byte[] password,
       int serverPort) {
     this.channel = channel;
     this.regionProvider = regionProvider;
     this.pubsub = pubsub;
     this.allowUnsupportedSupplier = allowUnsupportedSupplier;
     this.shutdownInvoker = shutdownInvoker;
     this.redisStats = redisStats;
     this.subscriberGroup = subscriberGroup;
     this.client = new Client(channel);
     this.byteBufAllocator = this.channel.alloc();
     this.authPassword = password;
     this.isAuthenticated = password == null;
     this.serverPort = serverPort;
     this.scanCursor = new BigInteger("0");
     this.sscanCursor = new BigInteger("0");
     this.hscanCursor = new BigInteger("0");
     redisStats.addClient();

     backgroundExecutor.submit(this::processCommandQueue);
   }

   public ChannelFuture writeToChannel(RedisResponse response) {
     return channel.writeAndFlush(response.encode(byteBufAllocator), channel.newPromise())
         .addListener((ChannelFutureListener) f -> {
           response.afterWrite();
           logResponse(response, channel.remoteAddress().toString(), f.cause());
         });
   }

   private void processCommandQueue() {
     while (true) {
       Command command = takeCommandFromQueue();
       if (command == TERMINATE_COMMAND) {
         return;
       }
       try {
         executeCommand(command);
         redisStats.incCommandsProcessed();
       } catch (Throwable ex) {
         exceptionCaught(command.getChannelHandlerContext(), ex);
       }
     }
   }

   private Command takeCommandFromQueue() {
     try {
       return commandQueue.take();
     } catch (InterruptedException e) {
       logger.info("Command queue thread interrupted");
       return TERMINATE_COMMAND;
     }
   }

   /**
    * This will handle the execution of received commands
    */
   @Override
   public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
     Command command = (Command) msg;
     command.setChannelHandlerContext(ctx);
     if (!channelInactive.get()) {
       commandQueue.put(command);
     }
   }

   /**
    * Exception handler for the entire pipeline
    */
   @Override
   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
     RedisResponse exceptionResponse = getExceptionResponse(ctx, cause);
     if (exceptionResponse != null) {
       writeToChannel(exceptionResponse);
     }
   }

   public EventLoopGroup getSubscriberGroup() {
     return subscriberGroup;
   }

   public synchronized void changeChannelEventLoopGroup(EventLoopGroup newGroup,
       Consumer<Boolean> callback) {
     if (newGroup.equals(channel.eventLoop())) {
       // already registered with newGroup
       callback.accept(true);
       return;
     }
     channel.deregister().addListener((ChannelFutureListener) future -> {
       boolean registerSuccess = true;
       synchronized (channel) {
         if (!channel.isRegistered()) {
           try {
             newGroup.register(channel).sync();
           } catch (Exception e) {
             logger.warn("Unable to register new EventLoopGroup: {}", e.getMessage());
             registerSuccess = false;
           }
         }
       }
       callback.accept(registerSuccess);
     });
   }

   private RedisResponse getExceptionResponse(ChannelHandlerContext ctx, Throwable cause) {
     RedisResponse response;
     if (cause instanceof IOException) {
       channelInactive(ctx);
       return null;
     }

     if (cause instanceof FunctionException
         && !(cause instanceof FunctionInvocationTargetException)) {
       Throwable th = CommandFunction.getInitialCause((FunctionException) cause);
       if (th != null) {
         cause = th;
       }
     }

     if (cause instanceof NumberFormatException) {
       response = RedisResponse.error(cause.getMessage());
     } else if (cause instanceof ArithmeticException) {
       response = RedisResponse.error(cause.getMessage());
     } else if (cause instanceof RedisDataTypeMismatchException) {
       response = RedisResponse.wrongType(cause.getMessage());
     } else if (cause instanceof DecoderException
         && cause.getCause() instanceof RedisCommandParserException) {
       response = RedisResponse.error(RedisConstants.PARSING_EXCEPTION_MESSAGE);

     } else if (cause instanceof InterruptedException || cause instanceof CacheClosedException) {
       response = RedisResponse.error(RedisConstants.SERVER_ERROR_SHUTDOWN);
     } else if (cause instanceof IllegalStateException
         || cause instanceof RedisParametersMismatchException) {
       response = RedisResponse.error(cause.getMessage());
     } else if (cause instanceof FunctionInvocationTargetException
         || cause instanceof DistributedSystemDisconnectedException
         || cause instanceof ForcedDisconnectException) {
       // This indicates a member departed or got disconnected
       logger.warn(
           "Closing client connection because one of the servers doing this operation departed.");
       channelInactive(ctx);
       response = null;
     } else {
       if (logger.isErrorEnabled()) {
         logger.error("GeodeRedisServer-Unexpected error handler for " + ctx.channel(), cause);
       }
       response = RedisResponse.error(RedisConstants.SERVER_ERROR_MESSAGE);
     }

     return response;
   }

   @Override
   public void channelInactive(ChannelHandlerContext ctx) {
     if (channelInactive.compareAndSet(false, true)) {
       if (logger.isDebugEnabled()) {
         logger.debug("GeodeRedisServer-Connection closing with " + ctx.channel().remoteAddress());
       }

       commandQueue.clear();
       commandQueue.offer(TERMINATE_COMMAND);
       redisStats.removeClient();
       ctx.channel().close();
       ctx.close();
     }
   }

   private void executeCommand(Command command) {
     try {
       if (logger.isDebugEnabled()) {
         logger.debug("Executing Redis command: {} - {}", command,
             channel.remoteAddress().toString());
       }

       if (command.isUnknown()) {
         writeToChannel(command.execute(this));
         return;
       }

       if (!isAuthenticated()) {
         writeToChannel(handleUnAuthenticatedCommand(command));
         return;
       }

       if (command.isUnsupported() && !allowUnsupportedCommands()) {
         writeToChannel(
             RedisResponse
                 .error(command.getCommandType() + RedisConstants.ERROR_UNSUPPORTED_COMMAND));
         return;
       }

       if (command.isUnimplemented()) {
         logger.info("Failed " + command.getCommandType() + " because it is not implemented.");
         writeToChannel(RedisResponse.error(command.getCommandType() + " is not implemented."));
         return;
       }

       if (!getPubSub().findSubscriptionNames(getClient()).isEmpty()) {
         if (!command.getCommandType().isAllowedWhileSubscribed()) {
           writeToChannel(RedisResponse
               .error("only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context"));
         }
       }

       final long start = redisStats.startCommand();
       try {
         writeToChannel(command.execute(this));
       } finally {
         redisStats.endCommand(command.getCommandType(), start);
       }

       if (command.isOfType(RedisCommandType.QUIT)) {
         channelInactive(command.getChannelHandlerContext());
       }
     } catch (Exception e) {
       logger.warn("Execution of Redis command {} failed: {}", command, e);
       throw e;
     }
   }

   private boolean allowUnsupportedCommands() {
     return allowUnsupportedSupplier.get();
   }

   private RedisResponse handleUnAuthenticatedCommand(Command command) {
     RedisResponse response;
     if (command.isOfType(RedisCommandType.AUTH)) {
       response = command.execute(this);
     } else {
       response = RedisResponse.customError(RedisConstants.ERROR_NOT_AUTH);
     }
     return response;
   }

   private void logResponse(RedisResponse response, String extraMessage, Throwable cause) {
     if (logger.isDebugEnabled() && response != null) {
       ByteBuf buf = response.encode(new UnpooledByteBufAllocator(false));
       if (cause == null) {
         logger.debug("Redis command returned: {} - {}",
             Command.getHexEncodedString(buf.array(), buf.readableBytes()), extraMessage);
       } else {
         logger.debug("Redis command FAILED to return: {} - {}",
             Command.getHexEncodedString(buf.array(), buf.readableBytes()), extraMessage, cause);
       }
     }
   }

   /**
    * {@link ByteBuf} allocator for this context. All executors must use this pooled allocator as
    * opposed to having unpooled buffers for maximum performance
    *
    * @return allocator instance
    */
   public ByteBufAllocator getByteBufAllocator() {
     return this.byteBufAllocator;
   }

   /**
    * Gets the provider of Regions
    */
   public RegionProvider getRegionProvider() {
     return regionProvider;
   }

   /**
    * Get the authentication password, this will be same server wide. It is exposed here as opposed
    * to {@link GeodeRedisServer}.
    */
   public byte[] getAuthPassword() {
     return this.authPassword;
   }

   /**
    * Checker if user has authenticated themselves
    *
    * @return True if no authentication required or authentication complete, false otherwise
    */
   public boolean isAuthenticated() {
     return this.isAuthenticated;
   }

   /**
    * Lets this context know the authentication is complete
    */
   public void setAuthenticationVerified() {
     this.isAuthenticated = true;
   }

   public int getServerPort() {
     return serverPort;
   }

   public Client getClient() {
     return client;
   }

   public void shutdown() {
     shutdownInvoker.run();
   }

   public PubSub getPubSub() {
     return pubsub;
   }

   public RedisStats getRedisStats() {
     return redisStats;
   }

   public BigInteger getScanCursor() {
     return scanCursor;
   }

   public void setScanCursor(BigInteger scanCursor) {
     this.scanCursor = scanCursor;
   }

   public BigInteger getSscanCursor() {
     return sscanCursor;
   }

   public void setSscanCursor(BigInteger sscanCursor) {
     this.sscanCursor = sscanCursor;
   }

   public BigInteger getHscanCursor() {
     return hscanCursor;
   }

   public void setHscanCursor(BigInteger hscanCursor) {
     this.hscanCursor = hscanCursor;
   }

   /**
    * This method and {@link #eventLoopReady()} are relevant for pubsub related commands which need
    * to return responses on a different EventLoopGroup. We need to ensure that the EventLoopGroup
    * switch has occurred before subsequent commands are executed.
    */
   public CountDownLatch getOrCreateEventLoopLatch() {
     if (eventLoopSwitched != null) {
       return eventLoopSwitched;
     }

     eventLoopSwitched = new CountDownLatch(1);
     return eventLoopSwitched;
   }

   /**
    * Signals that we have successfully switched over to a new EventLoopGroup.
    */
   public void eventLoopReady() {
     if (eventLoopSwitched == null) {
       return;
     }

     try {
       eventLoopSwitched.await();
     } catch (InterruptedException e) {
       logger.info("Event loop interrupted", e);
     }
   }
 }
	/*
	* 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.geode.redis.internal.netty;


	import java.io.IOException;
	import java.math.BigInteger;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.function.Consumer;
	import java.util.function.Supplier;

	import io.netty.buffer.ByteBuf;
	import io.netty.buffer.ByteBufAllocator;
	import io.netty.buffer.UnpooledByteBufAllocator;
	import io.netty.channel.Channel;
	import io.netty.channel.ChannelFuture;
	import io.netty.channel.ChannelFutureListener;
	import io.netty.channel.ChannelHandlerContext;
	import io.netty.channel.ChannelInboundHandlerAdapter;
	import io.netty.channel.EventLoopGroup;
	import io.netty.handler.codec.DecoderException;
	import org.apache.logging.log4j.Logger;

	import org.apache.geode.ForcedDisconnectException;
	import org.apache.geode.cache.CacheClosedException;
	import org.apache.geode.cache.execute.FunctionException;
	import org.apache.geode.cache.execute.FunctionInvocationTargetException;
	import org.apache.geode.distributed.DistributedSystemDisconnectedException;
	import org.apache.geode.logging.internal.log4j.api.LogService;
	import org.apache.geode.redis.internal.GeodeRedisServer;
	import org.apache.geode.redis.internal.ParameterRequirements.RedisParametersMismatchException;
	import org.apache.geode.redis.internal.RedisCommandType;
	import org.apache.geode.redis.internal.RedisConstants;
	import org.apache.geode.redis.internal.RegionProvider;
	import org.apache.geode.redis.internal.data.RedisDataTypeMismatchException;
	import org.apache.geode.redis.internal.executor.CommandFunction;
	import org.apache.geode.redis.internal.executor.RedisResponse;
	import org.apache.geode.redis.internal.pubsub.PubSub;
	import org.apache.geode.redis.internal.statistics.RedisStats;

	/**
	* This class extends {@link ChannelInboundHandlerAdapter} from Netty and it is the last part of the
	* channel pipeline. The {@link ByteToCommandDecoder} forwards a {@link Command} to this class which
	* executes it and sends the result back to the client. Additionally, all exception handling is done
	* by this class.
	* <p>
	* Besides being part of Netty's pipeline, this class also serves as a context to the execution of a
	* command. It provides access to the {@link RegionProvider} and anything else an executing {@link
	* Command} may need.
	*/
	public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {

	private static final Logger logger = LogService.getLogger();
	private static final Command TERMINATE_COMMAND = new Command();

	private final Client client;
	private final Channel channel;
	private final RegionProvider regionProvider;
	private final PubSub pubsub;
	private final ByteBufAllocator byteBufAllocator;
	private final byte[] authPassword;
	private final Supplier<Boolean> allowUnsupportedSupplier;
	private final Runnable shutdownInvoker;
	private final RedisStats redisStats;
	private final EventLoopGroup subscriberGroup;
	private BigInteger scanCursor;
	private BigInteger sscanCursor;
	private BigInteger hscanCursor;
	private final AtomicBoolean channelInactive = new AtomicBoolean();
	private final int MAX_QUEUED_COMMANDS =
	Integer.getInteger("geode.redis.commandQueueSize", 1000);
	private final LinkedBlockingQueue<Command> commandQueue =
	new LinkedBlockingQueue<>(MAX_QUEUED_COMMANDS);

	private final int serverPort;
	private CountDownLatch eventLoopSwitched;

	private boolean isAuthenticated;

	/**
	* Default constructor for execution contexts.
	*
	* @param channel Channel used by this context, should be one to one
	* @param password Authentication password for each context, can be null
	*/
	public ExecutionHandlerContext(Channel channel,
	RegionProvider regionProvider,
	PubSub pubsub,
	Supplier<Boolean> allowUnsupportedSupplier,
	Runnable shutdownInvoker,
	RedisStats redisStats,
	ExecutorService backgroundExecutor,
	EventLoopGroup subscriberGroup,
	byte[] password,
	int serverPort) {
	this.channel = channel;
	this.regionProvider = regionProvider;
	this.pubsub = pubsub;
	this.allowUnsupportedSupplier = allowUnsupportedSupplier;
	this.shutdownInvoker = shutdownInvoker;
	this.redisStats = redisStats;
	this.subscriberGroup = subscriberGroup;
	this.client = new Client(channel);
	this.byteBufAllocator = this.channel.alloc();
	this.authPassword = password;
	this.isAuthenticated = password == null;
	this.serverPort = serverPort;
	this.scanCursor = new BigInteger("0");
	this.sscanCursor = new BigInteger("0");
	this.hscanCursor = new BigInteger("0");
	redisStats.addClient();

	backgroundExecutor.submit(this::processCommandQueue);
	}

	public ChannelFuture writeToChannel(RedisResponse response) {
	return channel.writeAndFlush(response.encode(byteBufAllocator), channel.newPromise())
	.addListener((ChannelFutureListener) f -> {
	response.afterWrite();
	logResponse(response, channel.remoteAddress().toString(), f.cause());
	});
	}

	private void processCommandQueue() {
	while (true) {
	Command command = takeCommandFromQueue();
	if (command == TERMINATE_COMMAND) {
	return;
	}
	try {
	executeCommand(command);
	redisStats.incCommandsProcessed();
	} catch (Throwable ex) {
	exceptionCaught(command.getChannelHandlerContext(), ex);
	}
	}
	}

	private Command takeCommandFromQueue() {
	try {
	return commandQueue.take();
	} catch (InterruptedException e) {
	logger.info("Command queue thread interrupted");
	return TERMINATE_COMMAND;
	}
	}

	/**
	* This will handle the execution of received commands
	*/
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
	Command command = (Command) msg;
	command.setChannelHandlerContext(ctx);
	if (!channelInactive.get()) {
	commandQueue.put(command);
	}
	}

	/**
	* Exception handler for the entire pipeline
	*/
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
	RedisResponse exceptionResponse = getExceptionResponse(ctx, cause);
	if (exceptionResponse != null) {
	writeToChannel(exceptionResponse);
	}
	}

	public EventLoopGroup getSubscriberGroup() {
	return subscriberGroup;
	}

	public synchronized void changeChannelEventLoopGroup(EventLoopGroup newGroup,
	Consumer<Boolean> callback) {
	if (newGroup.equals(channel.eventLoop())) {
	// already registered with newGroup
	callback.accept(true);
	return;
	}
	channel.deregister().addListener((ChannelFutureListener) future -> {
	boolean registerSuccess = true;
	synchronized (channel) {
	if (!channel.isRegistered()) {
	try {
	newGroup.register(channel).sync();
	} catch (Exception e) {
	logger.warn("Unable to register new EventLoopGroup: {}", e.getMessage());
	registerSuccess = false;
	}
	}
	}
	callback.accept(registerSuccess);
	});
	}

	private RedisResponse getExceptionResponse(ChannelHandlerContext ctx, Throwable cause) {
	RedisResponse response;
	if (cause instanceof IOException) {
	channelInactive(ctx);
	return null;
	}

	if (cause instanceof FunctionException
	&& !(cause instanceof FunctionInvocationTargetException)) {
	Throwable th = CommandFunction.getInitialCause((FunctionException) cause);
	if (th != null) {
	cause = th;
	}
	}

	if (cause instanceof NumberFormatException) {
	response = RedisResponse.error(cause.getMessage());
	} else if (cause instanceof ArithmeticException) {
	response = RedisResponse.error(cause.getMessage());
	} else if (cause instanceof RedisDataTypeMismatchException) {
	response = RedisResponse.wrongType(cause.getMessage());
	} else if (cause instanceof DecoderException
	&& cause.getCause() instanceof RedisCommandParserException) {
	response = RedisResponse.error(RedisConstants.PARSING_EXCEPTION_MESSAGE);

	} else if (cause instanceof InterruptedException \|\| cause instanceof CacheClosedException) {
	response = RedisResponse.error(RedisConstants.SERVER_ERROR_SHUTDOWN);
	} else if (cause instanceof IllegalStateException
	\|\| cause instanceof RedisParametersMismatchException) {
	response = RedisResponse.error(cause.getMessage());
	} else if (cause instanceof FunctionInvocationTargetException
	\|\| cause instanceof DistributedSystemDisconnectedException
	\|\| cause instanceof ForcedDisconnectException) {
	// This indicates a member departed or got disconnected
	logger.warn(
	"Closing client connection because one of the servers doing this operation departed.");
	channelInactive(ctx);
	response = null;
	} else {
	if (logger.isErrorEnabled()) {
	logger.error("GeodeRedisServer-Unexpected error handler for " + ctx.channel(), cause);
	}
	response = RedisResponse.error(RedisConstants.SERVER_ERROR_MESSAGE);
	}

	return response;
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) {
	if (channelInactive.compareAndSet(false, true)) {
	if (logger.isDebugEnabled()) {
	logger.debug("GeodeRedisServer-Connection closing with " + ctx.channel().remoteAddress());
	}

	commandQueue.clear();
	commandQueue.offer(TERMINATE_COMMAND);
	redisStats.removeClient();
	ctx.channel().close();
	ctx.close();
	}
	}

	private void executeCommand(Command command) {
	try {
	if (logger.isDebugEnabled()) {
	logger.debug("Executing Redis command: {} - {}", command,
	channel.remoteAddress().toString());
	}

	if (command.isUnknown()) {
	writeToChannel(command.execute(this));
	return;
	}

	if (!isAuthenticated()) {
	writeToChannel(handleUnAuthenticatedCommand(command));
	return;
	}

	if (command.isUnsupported() && !allowUnsupportedCommands()) {
	writeToChannel(
	RedisResponse
	.error(command.getCommandType() + RedisConstants.ERROR_UNSUPPORTED_COMMAND));
	return;
	}

	if (command.isUnimplemented()) {
	logger.info("Failed " + command.getCommandType() + " because it is not implemented.");
	writeToChannel(RedisResponse.error(command.getCommandType() + " is not implemented."));
	return;
	}

	if (!getPubSub().findSubscriptionNames(getClient()).isEmpty()) {
	if (!command.getCommandType().isAllowedWhileSubscribed()) {
	writeToChannel(RedisResponse
	.error("only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context"));
	}
	}

	final long start = redisStats.startCommand();
	try {
	writeToChannel(command.execute(this));
	} finally {
	redisStats.endCommand(command.getCommandType(), start);
	}

	if (command.isOfType(RedisCommandType.QUIT)) {
	channelInactive(command.getChannelHandlerContext());
	}
	} catch (Exception e) {
	logger.warn("Execution of Redis command {} failed: {}", command, e);
	throw e;
	}
	}

	private boolean allowUnsupportedCommands() {
	return allowUnsupportedSupplier.get();
	}

	private RedisResponse handleUnAuthenticatedCommand(Command command) {
	RedisResponse response;
	if (command.isOfType(RedisCommandType.AUTH)) {
	response = command.execute(this);
	} else {
	response = RedisResponse.customError(RedisConstants.ERROR_NOT_AUTH);
	}
	return response;
	}

	private void logResponse(RedisResponse response, String extraMessage, Throwable cause) {
	if (logger.isDebugEnabled() && response != null) {
	ByteBuf buf = response.encode(new UnpooledByteBufAllocator(false));
	if (cause == null) {
	logger.debug("Redis command returned: {} - {}",
	Command.getHexEncodedString(buf.array(), buf.readableBytes()), extraMessage);
	} else {
	logger.debug("Redis command FAILED to return: {} - {}",
	Command.getHexEncodedString(buf.array(), buf.readableBytes()), extraMessage, cause);
	}
	}
	}

	/**
	* {@link ByteBuf} allocator for this context. All executors must use this pooled allocator as
	* opposed to having unpooled buffers for maximum performance
	*
	* @return allocator instance
	*/
	public ByteBufAllocator getByteBufAllocator() {
	return this.byteBufAllocator;
	}

	/**
	* Gets the provider of Regions
	*/
	public RegionProvider getRegionProvider() {
	return regionProvider;
	}

	/**
	* Get the authentication password, this will be same server wide. It is exposed here as opposed
	* to {@link GeodeRedisServer}.
	*/
	public byte[] getAuthPassword() {
	return this.authPassword;
	}

	/**
	* Checker if user has authenticated themselves
	*
	* @return True if no authentication required or authentication complete, false otherwise
	*/
	public boolean isAuthenticated() {
	return this.isAuthenticated;
	}

	/**
	* Lets this context know the authentication is complete
	*/
	public void setAuthenticationVerified() {
	this.isAuthenticated = true;
	}

	public int getServerPort() {
	return serverPort;
	}

	public Client getClient() {
	return client;
	}

	public void shutdown() {
	shutdownInvoker.run();
	}

	public PubSub getPubSub() {
	return pubsub;
	}

	public RedisStats getRedisStats() {
	return redisStats;
	}

	public BigInteger getScanCursor() {
	return scanCursor;
	}

	public void setScanCursor(BigInteger scanCursor) {
	this.scanCursor = scanCursor;
	}

	public BigInteger getSscanCursor() {
	return sscanCursor;
	}

	public void setSscanCursor(BigInteger sscanCursor) {
	this.sscanCursor = sscanCursor;
	}

	public BigInteger getHscanCursor() {
	return hscanCursor;
	}

	public void setHscanCursor(BigInteger hscanCursor) {
	this.hscanCursor = hscanCursor;
	}

	/**
	* This method and {@link #eventLoopReady()} are relevant for pubsub related commands which need
	* to return responses on a different EventLoopGroup. We need to ensure that the EventLoopGroup
	* switch has occurred before subsequent commands are executed.
	*/
	public CountDownLatch getOrCreateEventLoopLatch() {
	if (eventLoopSwitched != null) {
	return eventLoopSwitched;
	}

	eventLoopSwitched = new CountDownLatch(1);
	return eventLoopSwitched;
	}

	/**
	* Signals that we have successfully switched over to a new EventLoopGroup.
	*/
	public void eventLoopReady() {
	if (eventLoopSwitched == null) {
	return;
	}

	try {
	eventLoopSwitched.await();
	} catch (InterruptedException e) {
	logger.info("Event loop interrupted", e);
	}
	}
	}