content: /*
   * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
   *
   * 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.otter.common.push.supplier.media;
  
  import java.net.InetSocketAddress;
  import java.util.HashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.apache.commons.lang.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.common.push.PushException;
  import com.alibaba.otter.common.push.SubscribeCallback;
  import com.alibaba.otter.common.push.SubscribeManager;
  import com.alibaba.otter.common.push.SubscribeManagerFactory;
  import com.alibaba.otter.common.push.SubscribeType;
  import com.alibaba.otter.common.push.supplier.AbstractDatasourceSupplier;
  import com.alibaba.otter.common.push.supplier.DatasourceChangeCallback;
  import com.alibaba.otter.common.push.supplier.DatasourceInfo;
  import com.alibaba.otter.common.push.supplier.HaDatasourceInfo;
  import com.alibaba.otter.shared.common.utils.JsonUtils;
  
  /**
   * 基于data media机制提供的数据源信息
   * 
   * @author jianghang 2013-4-18 下午03:03:26
   * @version 4.1.8
   */
  public class MediaDatasourceSupplier extends AbstractDatasourceSupplier {
  
      private static final Logger            log          = LoggerFactory.getLogger(MediaDatasourceSupplier.class);
      private static AtomicInteger           CHANGED_TIME = new AtomicInteger(0);
  
      private SubscribeManager               mediaSubscribeManager;
  
      private String                         groupKey;
      private HaDatasourceInfo               haInfo;
  
      private List<DatasourceChangeCallback> callbacks    = new LinkedList<DatasourceChangeCallback>();
  
      private Object                         lock         = new Object();
      private SubscribeCallback              subscribeCallback;
  
      private MediaDatasourceSupplier(){
      }
  
      private MediaDatasourceSupplier(String groupKey){
          this.groupKey = groupKey;
      }
  
      public static MediaDatasourceSupplier newInstance(String groupKey) {
          MediaDatasourceSupplier supplier = new MediaDatasourceSupplier(groupKey);
          return supplier;
      }
  
      public void doStart() {
          subscribeCallback = new SubscribeCallback() {
  
              @Override
              public void callback(String matrixInfo) {
  
                  log.warn("has received changed ds [{}] for [{}] times", matrixInfo, CHANGED_TIME.addAndGet(1));
                  synchronized (MediaDatasourceSupplier.this.lock) {
                      MediaDatasourceSupplier.this.haInfo = parse(matrixInfo);
                      MediaDatasourceSupplier.this.callback();
                  }
              }
          };
          this.init();
      }
  
      @Override
      public void doStop() {
          this.mediaSubscribeManager.unRegisterCallback(this.groupKey, subscribeCallback);
          this.callbacks.clear();
      }
  
      public synchronized void init() {
          this.mediaSubscribeManager = SubscribeManagerFactory.getSubscribeManager(SubscribeType.MEDIA);
          if (this.mediaSubscribeManager == null) {
              throw new PushException("MediaDatasourceSupplier : mediaSubscribeManager is null, check the spring config");
          }
  
          String matrixStr = mediaSubscribeManager.fetchConfig(groupKey);
          this.haInfo = parse(matrixStr);
          this.mediaSubscribeManager.registerCallback(this.groupKey, subscribeCallback);
      }
  
      public DatasourceInfo fetchMaster() {
          synchronized (lock) {
              if (this.haInfo == null) {
                  throw new PushException("haInfo is null, check the init phase");
              }
              return this.haInfo.getMaster();
          }
      }
  
      public void addSwtichCallback(DatasourceChangeCallback callback) {
          callbacks.add(callback);
      }
  
      private void callback() {
          if (callbacks == null || callbacks.size() == 0) {
              return;
          }
  
          for (DatasourceChangeCallback callback : callbacks) {
              callback.masterChanged(MediaDatasourceSupplier.this.fetchMaster());
          }
  
      }
  
      private HaDatasourceInfo parse(String matrixStr) {
          HaDatasourceInfo haInfo = new HaDatasourceInfo();
  
          Map jsonMap = JsonUtils.unmarshalFromString(matrixStr, HashMap.class);
          String masterAddress = (String) jsonMap.get("master");
          if (masterAddress != null) {
              DatasourceInfo master = new DatasourceInfo();
              master.setAddress(parseAddress(masterAddress));
              haInfo.setMaster(master);
          }
  
          String slaveAddress = (String) jsonMap.get("slave");
          if (slaveAddress != null) {
              DatasourceInfo slave = new DatasourceInfo();
              slave.setAddress(parseAddress(slaveAddress));
              haInfo.getSlavers().add(slave);
          }
  
          return haInfo;
      }
  
      private InetSocketAddress parseAddress(String address) {
          String[] strs = StringUtils.split(address, ":");
          if (strs.length != 2) {
              throw new IllegalArgumentException("illegal address format:" + address);
          }
  
          return new InetSocketAddress(strs[0], Integer.valueOf(strs[1]));
      }
  
  }
  