<commit_before>/**
 * Copyright (C) 2010-2011, FuseSource Corp.  All rights reserved.
 *
 *     http://fusesource.com
 *
 * The software in this package is published under the terms of the
 * CDDL license a copy of which has been included with this distribution
 * in the license.txt file.
 */
package org.fusesource.fabric.groups


import collection.mutable.{ListBuffer, HashMap}
import internal.ChangeListenerSupport

import java.io._
import org.codehaus.jackson.map.ObjectMapper
import collection.JavaConversions._
import java.util.LinkedHashMap
import java.lang.{IllegalStateException, String}

/**
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
trait NodeState {

  /**
   * The id of the cluster node.  There can be multiple node with this ID,
   * but only the first node in the cluster will be the master for for it.
   */
  def id: String
}

/**
 *
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
object ClusteredSupport {

  private var mapper: ObjectMapper = new ObjectMapper

  def decode[T](t : Class[T], buffer: Array[Byte]): T = decode(t, new ByteArrayInputStream(buffer))
  def decode[T](t : Class[T], in: InputStream): T =  mapper.readValue(in, t)

  def encode(value: AnyRef): Array[Byte] = {
    var baos: ByteArrayOutputStream = new ByteArrayOutputStream
    encode(value, baos)
    return baos.toByteArray
  }

  def encode(value: AnyRef, out: OutputStream): Unit = {
    mapper.writeValue(out, value)
  }

}

/**
 * <p>
 * </p>
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
class ClusteredSingletonWatcher[T <: NodeState](val stateClass:Class[T]) extends ChangeListenerSupport {
  import ClusteredSupport._
  
  protected var _group:Group = _
  def group = _group

  private val listener = new ChangeListener() {
    def changed() {
      val members = _group.members
      val t = new LinkedHashMap[String, T]()
      members.foreach {
        case (path, data) =>
          try {
            val value = decode(stateClass, data)
            t.put(path, value)
          } catch {
            case e: Throwable =>
              e.printStackTrace()
          }
      }
      changed_decoded(t)
    }

    def connected = ClusteredSingletonWatcher.this.fireConnected
    def disconnected = ClusteredSingletonWatcher.this.fireDisconnected
  }


  def start(group:Group) = this.synchronized {
    if(_group !=null )
      throw new IllegalStateException("Already started.")
    _group = group
    _group.add(listener)
  }

  def stop = this.synchronized {
    if(_group==null)
      throw new IllegalStateException("Not started.")
    _group.remove(listener)
    _members = HashMap[String, ListBuffer[(String,  T)]]()
    _group = null
  }

  def connected = this.synchronized {
    if(_group==null) {
      false
    } else {
      _group.connected
    }
  }

  protected var _members = HashMap[String, ListBuffer[(String,  T)]]()
  def members = this.synchronized { _members }

  def changed_decoded(m: LinkedHashMap[String, T]) = {
    this.synchronized {
      if( _group!=null ) {
        _members = HashMap[String, ListBuffer[(String,  T)]]()
        m.foreach { case node =>
          _members.getOrElseUpdate(node._2.id, ListBuffer[(String,  T)]()).append(node)
        }
      }
    }
    fireChanged
  }

  def masters = this.synchronized {
    _members.mapValues(_.head._2).toArray.map(_._2)
  }

}
/**
 * <p>
 * </p>
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
class ClusteredSingleton[T <: NodeState](stateClass:Class[T], val id:String) extends ClusteredSingletonWatcher[T](stateClass) {
  import ClusteredSupport._

  private var _eid:String = _<commit_msg>Update the masters method to return an array of master nodes in the cluster (instead of seq)<commit_after>/**
 * Copyright (C) 2010-2011, FuseSource Corp.  All rights reserved.
 *
 *     http://fusesource.com
 *
 * The software in this package is published under the terms of the
 * CDDL license a copy of which has been included with this distribution
 * in the license.txt file.
 */
package org.fusesource.fabric.groups


import collection.mutable.{ListBuffer, HashMap}
import internal.ChangeListenerSupport

import java.io._
import org.codehaus.jackson.map.ObjectMapper
import collection.JavaConversions._
import java.util.LinkedHashMap
import java.lang.{IllegalStateException, String}

/**
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
trait NodeState {

  /**
   * The id of the cluster node.  There can be multiple node with this ID,
   * but only the first node in the cluster will be the master for for it.
   */
  def id: String
}

/**
 *
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
object ClusteredSupport {

  private var mapper: ObjectMapper = new ObjectMapper

  def decode[T](t : Class[T], buffer: Array[Byte]): T = decode(t, new ByteArrayInputStream(buffer))
  def decode[T](t : Class[T], in: InputStream): T =  mapper.readValue(in, t)

  def encode(value: AnyRef): Array[Byte] = {
    var baos: ByteArrayOutputStream = new ByteArrayOutputStream
    encode(value, baos)
    return baos.toByteArray
  }

  def encode(value: AnyRef, out: OutputStream): Unit = {
    mapper.writeValue(out, value)
  }

}

/**
 * <p>
 * </p>
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
class ClusteredSingletonWatcher[T <: NodeState](val stateClass:Class[T]) extends ChangeListenerSupport {
  import ClusteredSupport._
  
  protected var _group:Group = _
  def group = _group

  private val listener = new ChangeListener() {
    def changed() {
      val members = _group.members
      val t = new LinkedHashMap[String, T]()
      members.foreach {
        case (path, data) =>
          try {
            val value = decode(stateClass, data)
            t.put(path, value)
          } catch {
            case e: Throwable =>
              e.printStackTrace()
          }
      }
      changed_decoded(t)
    }

    def connected = ClusteredSingletonWatcher.this.fireConnected
    def disconnected = ClusteredSingletonWatcher.this.fireDisconnected
  }


  def start(group:Group) = this.synchronized {
    if(_group !=null )
      throw new IllegalStateException("Already started.")
    _group = group
    _group.add(listener)
  }

  def stop = this.synchronized {
    if(_group==null)
      throw new IllegalStateException("Not started.")
    _group.remove(listener)
    _members = HashMap[String, ListBuffer[(String,  T)]]()
    _group = null
  }

  def connected = this.synchronized {
    if(_group==null) {
      false
    } else {
      _group.connected
    }
  }

  protected var _members = HashMap[String, ListBuffer[(String,  T)]]()
  def members = this.synchronized { _members }

  def changed_decoded(m: LinkedHashMap[String, T]) = {
    this.synchronized {
      if( _group!=null ) {
        _members = HashMap[String, ListBuffer[(String,  T)]]()
        m.foreach { case node =>
          _members.getOrElseUpdate(node._2.id, ListBuffer[(String,  T)]()).append(node)
        }
      }
    }
    fireChanged
  }

  def masters = this.synchronized {
    _members.mapValues(_.head._2).toArray.map(_._2).toArray(new ClassManifest[T] {
      def erasure = stateClass
    })
  }

}
/**
 * <p>
 * </p>
 *
 * @author <a href="http://hiramchirino.com"><NAME></a>
 */
class ClusteredSingleton[T <: NodeState ](stateClass:Class[T], val id:String) extends ClusteredSingletonWatcher[T](stateClass) {
  import ClusteredSupport._

  private var _eid:String = _
{'fraction_non_alphanumeric': 0.1236422153681105, 'fraction_numerical': 0.0029502480890438512, 'mean_word_length': 3.115894039735099, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4247964', 'n_tokens_mistral': 2390, 'n_tokens_neox': 2335, 'n_words': 585}