content: /**
   * 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 com.alibaba.wasp.messagequeue;
  
  import com.alibaba.wasp.EntityGroupInfo;
  import com.alibaba.wasp.FConstants;
  import com.alibaba.wasp.fserver.EntityGroup;
  import com.alibaba.wasp.fserver.LeaseException;
  import com.alibaba.wasp.fserver.LeaseListener;
  import com.alibaba.wasp.fserver.Leases;
  import com.alibaba.wasp.fserver.Leases.LeaseStillHeldException;
  import com.alibaba.wasp.fserver.OnlineEntityGroups;
  import com.alibaba.wasp.storage.StorageTableNotFoundException;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hbase.HColumnDescriptor;
  import org.apache.hadoop.hbase.HTableDescriptor;
  import org.apache.hadoop.hbase.client.HBaseAdmin;
  import org.apache.hadoop.hbase.io.hfile.Compression;
  
  import java.io.Closeable;
  import java.io.IOException;
  import java.util.Collection;
  import java.util.Iterator;
  import java.util.List;
  import java.util.concurrent.ConcurrentHashMap;
  
  /**
   * Message broker, it receives message from message queue and send message to
   * subscriber.
   * 
   */
  public class MessageBroker extends Thread implements Closeable, Broker {
  
    public static final Log LOG = LogFactory.getLog(MessageBroker.class);
  
    /** entityGroup server **/
    private final OnlineEntityGroups service;
  
    /** for register **/
    protected final ConcurrentHashMap<String, Subscriber> subscribers = new ConcurrentHashMap<String, Subscriber>();
  
    private boolean closed = false;
    private Leases leases;
    private final int subscriberLeaseTimeoutPeriod;
    private final RenewRunnable renew;
  
    private final Configuration conf;
  
    /**
     * @param service
     *          entityGroup server instance
     * @param conf
     *          configuration
     */
    public MessageBroker(OnlineEntityGroups service, Configuration conf) {
      this.service = service;
      this.conf = conf;
      this.leases = new Leases(conf.getInt(FConstants.THREAD_WAKE_FREQUENCY,
          10 * 1000));
      this.subscriberLeaseTimeoutPeriod = conf.getInt(
          FConstants.WASP_FSEVER_SUBSCRIBER_TIMEOUT_PERIOD,
          FConstants.DEFAULT_WASP_FSEVER_SUBSCRIBER_TIMEOUT_PERIOD);
      this.renew = new RenewRunnable();
      this.renew.start();
    }
  
    public void initlize() throws IOException {
      HBaseAdmin admin = new HBaseAdmin(this.conf);
      if (!admin.tableExists(FConstants.MESSAGEQUEUE_TABLENAME)) {
        HColumnDescriptor family = new HColumnDescriptor(FConstants.MESSAGEQUEUE_FAMILIY);
        family.setCompressionType(Compression.Algorithm.GZ);
        HTableDescriptor tableDes = new HTableDescriptor(FConstants.MESSAGEQUEUE_TABLENAME);
        tableDes.addFamily(family);
        admin.createTable(tableDes);
      }
      admin.close();
    }
  
    private class SubscriberListener implements LeaseListener {
  
      private Subscriber subscriber;
  
      /**
       * @param subscriber
       */
      public SubscriberListener(Subscriber subscriber) {
        super();
        this.subscriber = subscriber;
      }
  
      /**
       * @see com.alibaba.wasp.fserver.LeaseListener#leaseExpired()
       */
      @Override
      public void leaseExpired() {
        Subscriber s = subscribers.remove(this.subscriber.getEntityGroup()
            .getEntityGroupNameAsString());
        LOG.info("Subscriber " + s.getEntityGroup().getEntityGroupNameAsString()
            + " lease expired");
      }
    }
  
    private class RenewRunnable extends Thread implements Closeable {
      private boolean closed = false;
  
      @Override
      public void run() {
  
        while (!closed) {
          // renew lease,if some entityGroups hang then remove it by lease
          // manager.
          Collection<EntityGroup> entityGroups;
          try {
            entityGroups = service.getOnlineEntityGroups();
            for (EntityGroup entityGroup : entityGroups) {
              leases.renewLease(entityGroup.getEntityGroupNameAsString());
            }
            Thread.sleep(60 * 1000);
          } catch (Exception e) {
            LOG.error("RenewRunnable running.", e);
            try {
              close();
            } catch (IOException e1) {
              LOG.error("RenewRunnable doClosing.", e1);
            }
            return;
          }
        }
      }
  
      /**
       * @see java.io.Closeable#close()
       */
      @Override
      public void close() throws IOException {
        closed = true;
      }
    }
  
    /**
     * @see
     */
    @Override
    public void register(Subscriber subscriber) throws LeaseStillHeldException {
      String subscriberName = subscriber.getEntityGroup()
          .getEntityGroupNameAsString();
      Subscriber existing = subscribers.putIfAbsent(subscriberName, subscriber);
      if (existing == null) {
        this.leases
            .createLease(subscriberName, this.subscriberLeaseTimeoutPeriod,
                new SubscriberListener(subscriber));
      }
    }
  
    /**
     * @see
     */
    @Override
    public void remove(Subscriber subscriber) throws LeaseException {
      String sbuscriberName = subscriber.getEntityGroup()
          .getEntityGroupNameAsString();
      subscribers.remove(sbuscriberName);
      leases.removeLease(sbuscriberName);
    }
  
    /**
     * @see
     */
    @Override
    public void remove(EntityGroupInfo entityGroupInfo) throws LeaseException {
      String sbuscriberName = entityGroupInfo.getEntityGroupNameAsString();
      subscribers.remove(sbuscriberName);
      leases.removeLease(sbuscriberName);
    }
  
    /**
     * @see Thread#run()
     */
    @Override
    public void run() {
      while (!closed) {
        try {
          // fetch message from message queue then notify entityGroup.
          selectCurrentMessages();
  
          // wait for 500ms
          Thread.sleep(500);
        } catch (Exception e) {
          LOG.error("Messagebroker running.", e);
          try {
            close();
          } catch (IOException e1) {
            LOG.error("Messagebroker doClosing.", e1);
          }
          return;
        }
      }
    }
  
    /**
     * Fetch messages which were subscribed.
     *
     * @throws HBaseTableNotFoundException
     * @throws java.io.IOException
     */
    private void selectCurrentMessages() throws StorageTableNotFoundException,
        IOException {
      Iterator<Subscriber> iter = subscribers.values().iterator();
      while (iter.hasNext()) {
        Subscriber subscriber = iter.next();
  
        List<Message> messages = subscriber.receive();
        for (Message message : messages) {
          subscriber.setCurrentMessageRow(message.getMessageID().getMessageId());
          subscriber.doAsynchronous(message);
        }
      }
    }
  
    @Override
    public void close() throws IOException {
      this.closed = true;
      this.renew.close();
      this.interrupt();
      this.renew.interrupt();
    }
  }