content: /*
   * Copyright 1999-2012 Alibaba Group.
   *  
   * Licensed 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 com.alibaba.cobar.net.handler;
  
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.Executor;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  import com.alibaba.cobar.net.NIOHandler;
  
  /**
   * @author xianmao.hexm 2012-7-13
   */
  public abstract class BackendAsyncHandler implements NIOHandler {
  
      protected final BlockingQueue<byte[]> dataQueue = new LinkedBlockingQueue<byte[]>();
      protected final AtomicBoolean isHandling = new AtomicBoolean(false);
  
      protected void offerData(byte[] data, Executor executor) {
          if (dataQueue.offer(data)) {
              handleQueue(executor);
          } else {
              offerDataError();
          }
      }
  
      protected abstract void offerDataError();
  
      protected abstract void handleData(byte[] data);
  
      protected abstract void handleDataError(Throwable t);
  
      protected void handleQueue(final Executor executor) {
          if (isHandling.compareAndSet(false, true)) {
              // 异步处理后端数据
              executor.execute(new Runnable() {
                  @Override
                  public void run() {
                      try {
                          byte[] data = null;
                          while ((data = dataQueue.poll()) != null) {
                              handleData(data);
                          }
                      } catch (Throwable t) {
                          handleDataError(t);
                      } finally {
                          isHandling.set(false);
                          if (dataQueue.size() > 0) {
                              handleQueue(executor);
                          }
                      }
                  }
              });
          }
      }
  
  }
  