text:
package kafka.network

import java.io._
import java.net._
import java.nio.ByteBuffer
import java.util.{HashMap, Random}
import java.nio.channels.SocketChannel
import javax.net.ssl._

import com.yammer.metrics.core.{Gauge, Meter}
import com.yammer.metrics.{Metrics => YammerMetrics}
import kafka.network.RequestChannel.SendAction
import kafka.security.CredentialProvider
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.memory.MemoryPool
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.{ChannelBuilder, ChannelState, KafkaChannel, ListenerName, NetworkReceive, NetworkSend, Selector, Send}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.MemoryRecords
import org.apache.kafka.common.requests.{AbstractRequest, ProduceRequest, RequestHeader}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.security.scram.internal.ScramMechanism
import org.apache.kafka.common.utils.{LogContext, MockTime, Time}
import org.apache.log4j.Level
import org.junit.Assert._
import org.junit._
import org.scalatest.junit.JUnitSuite

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.control.ControlThrowable

class SocketServerTest extends JUnitSuite {
  val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
  props.put("listeners", "PLAINTEXT://localhost:0")
  props.put("num.network.threads", "1")
  props.put("socket.send.buffer.bytes", "300000")
  props.put("socket.receive.buffer.bytes", "300000")
  props.put("queued.max.requests", "50")
  props.put("socket.request.max.bytes", "50")
  props.put("max.connections.per.ip", "5")
  props.put("connections.max.idle.ms", "60000")
  val config = KafkaConfig.fromProps(props)
  val metrics = new Metrics
  val credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, null)
  val localAddress = InetAddress.getLoopbackAddress

  // Clean-up any metrics left around by previous tests
  for (metricName <- YammerMetrics.defaultRegistry.allMetrics.keySet.asScala)
    YammerMetrics.defaultRegistry.removeMetric(metricName)

  val server = new SocketServer(config, metrics, Time.SYSTEM, credentialProvider)
  server.startup()
  val sockets = new ArrayBuffer[Socket]

  private val kafkaLogger = org.apache.log4j.LogManager.getLogger("kafka")
  private var logLevelToRestore: Level = _

  @Before
  def setUp(): Unit = {
    // Run the tests with TRACE logging to exercise request logging path
    logLevelToRestore = kafkaLogger.getLevel
    kafkaLogger.setLevel(Level.TRACE)
  }

  @After
  def tearDown() {
    shutdownServerAndMetrics(server)
    sockets.foreach(_.close())
    sockets.clear()
    kafkaLogger.setLevel(logLevelToRestore)
  }

  def sendRequest(socket: Socket, request: Array[Byte], id: Option[Short] = None, flush: Boolean = true) {
    val outgoing = new DataOutputStream(socket.getOutputStream)
    id match {
      case Some(id) =>
        outgoing.writeInt(request.length + 2)
        outgoing.writeShort(id)
      case None =>
        outgoing.writeInt(request.length)
    }
    outgoing.write(request)
    if (flush)
      outgoing.flush()
  }

  def receiveResponse(socket: Socket): Array[Byte] = {
    val incoming = new DataInputStream(socket.getInputStream)
    val len = incoming.readInt()
    val response = new Array[Byte](len)
    incoming.readFully(response)
    response
  }

  private def receiveRequest(channel: RequestChannel, timeout: Long = 2000L): RequestChannel.Request = {
    channel.receiveRequest(timeout) match {
      case request: RequestChannel.Request => request
      case RequestChannel.ShutdownRequest => fail("Unexpected shutdown received")
      case null => fail("receiveRequest timed out")
    }
  }

  /* A simple request handler that just echos back the response */
  def processRequest(channel: RequestChannel) {
    processRequest(channel, receiveRequest(channel))
  }

  def processRequest(channel: RequestChannel, request: RequestChannel.Request) {
    val byteBuffer = request.body[AbstractRequest].serialize(request.header)
    byteBuffer.rewind()

    val send = new NetworkSend(request.context.connectionId, byteBuffer)
    channel.sendResponse(new RequestChannel.Response(request, Some(send), SendAction, Some(request.header.toString)))
  }

  def connect(s: SocketServer = server, protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT, localAddr: InetAddress = null) = {
    val socket = new Socket("localhost", s.boundPort(ListenerName.forSecurityProtocol(protocol)), localAddr, 0)
    sockets += socket
    socket
  }

  // Create a client connection, process one request and return (client socket, connectionId)
  def connectAndProcessRequest(s: SocketServer): (Socket, String) = {
    val socket = connect(s)
    val request = sendAndReceiveRequest(socket, s)
    processRequest(s.requestChannel, request)
    (socket, request.context.connectionId)
  }

  def sendAndReceiveRequest(socket: Socket, server: SocketServer): RequestChannel.Request = {
    sendRequest(socket, producerRequestBytes())
    receiveRequest(server.requestChannel)
  }

  def shutdownServerAndMetrics(server: SocketServer): Unit = {
    server.shutdown()
    server.metrics.close()
  }

  private def producerRequestBytes(ack: Short = 0): Array[Byte] = {
    val correlationId = -1
    val clientId = ""
    val ackTimeoutMs = 10000

    val emptyRequest = ProduceRequest.Builder.forCurrentMagic(ack, ackTimeoutMs,
      new HashMap[TopicPartition, MemoryRecords]()).build()
    val emptyHeader = new RequestHeader(ApiKeys.PRODUCE, emptyRequest.version, clientId, correlationId)
    val byteBuffer = emptyRequest.serialize(emptyHeader)
    byteBuffer.rewind()

    val serializedBytes = new Array[Byte](byteBuffer.remaining)
    byteBuffer.get(serializedBytes)
    serializedBytes
  }

  @Test
  def simpleRequest() {
    val plainSocket = connect(protocol = SecurityProtocol.PLAINTEXT)
    val serializedBytes = producerRequestBytes()

    // Test PLAINTEXT socket
    sendRequest(plainSocket, serializedBytes)
    processRequest(server.requestChannel)
    assertEquals(serializedBytes.toSeq, receiveResponse(plainSocket).toSeq)
  }

  @Test
  def tooBigRequestIsRejected() {
    val tooManyBytes = new Array[Byte](server.config.socketRequestMaxBytes + 1)
    new Random().nextBytes(tooManyBytes)
    val socket = connect()
    val outgoing = new DataOutputStream(socket.getOutputStream)
    outgoing.writeInt(tooManyBytes.length)
    try {
      // Server closes client connection when it processes the request length because
      // it is too big. The write of request body may fail if the connection has been closed.
      outgoing.write(tooManyBytes)
      outgoing.flush()
      receiveResponse(socket)
    } catch {
      case _: IOException => // thats fine
    }
  }

  @Test
  def testGracefulClose() {
    val plainSocket = connect(protocol = SecurityProtocol.PLAINTEXT)
    val serializedBytes = producerRequestBytes()

    for (_ <- 0 until 10)
      sendRequest(plainSocket, serializedBytes)
    plainSocket.close()
    for (_ <- 0 until 10) {
      val request = receiveRequest(server.requestChannel)
      assertNotNull("receiveRequest timed out", request)
      server.requestChannel.sendResponse(new RequestChannel.Response(request, None, RequestChannel.NoOpAction, None))
    }
  }

  @Test
  def testNoOpAction(): Unit = {
    val plainSocket = connect(protocol = SecurityProtocol.PLAINTEXT)
    val serializedBytes = producerRequestBytes()

    for (_ <- 0 until 3)
      sendRequest(plainSocket, serializedBytes)
    for (_ <- 0 until 3) {
      val request = receiveRequest(server.requestChannel)
      assertNotNull("receiveRequest timed out", request)
      server.requestChannel.sendResponse(new RequestChannel.Response(request, None, RequestChannel.NoOpAction, None))
    }
  }

  @Test
  def testConnectionId() {
    val sockets = (1 to 5).map(_ => connect(protocol = SecurityProtocol.PLAINTEXT))
    val serializedBytes = producerRequestBytes()

    val requests = sockets.map{socket =>
      sendRequest(socket, serializedBytes)
      receiveRequest(server.requestChannel)
    }
    requests.zipWithIndex.foreach { case (request, i) =>
      val index = request.context.connectionId.split("-").last
      assertEquals(i.toString, index)
    }

    sockets.foreach(_.close)
  }

  @Test
  def testIdleConnection() {
    val idleTimeMs = 60000
    val time = new MockTime()
    props.put(KafkaConfig.ConnectionsMaxIdleMsProp, idleTimeMs.toString)
    val serverMetrics = new Metrics
    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, time, credentialProvider)

    def openChannel(request: RequestChannel.Request): Option[KafkaChannel] =
      overrideServer.processor(request.processor).channel(request.context.connectionId)
    def openOrClosingChannel(request: RequestChannel.Request): Option[KafkaChannel] =
      overrideServer.processor(request.processor).openOrClosingChannel(request.context.connectionId)

    try {
      overrideServer.startup()
      val serializedBytes = producerRequestBytes()

      // Connection with no staged receives
      val socket1 = connect(overrideServer, protocol = SecurityProtocol.PLAINTEXT)
      sendRequest(socket1, serializedBytes)
      val request1 = receiveRequest(overrideServer.requestChannel)
      assertTrue("Channel not open", openChannel(request1).nonEmpty)
      assertEquals(openChannel(request1), openOrClosingChannel(request1))

      time.sleep(idleTimeMs + 1)
      TestUtils.waitUntilTrue(() => openOrClosingChannel(request1).isEmpty, "Failed to close idle channel")
      assertTrue("Channel not removed", openChannel(request1).isEmpty)
      processRequest(overrideServer.requestChannel, request1)

      // Connection with staged receives
      val socket2 = connect(overrideServer, protocol = SecurityProtocol.PLAINTEXT)
      val request2 = sendRequestsUntilStagedReceive(overrideServer, socket2, serializedBytes)

      time.sleep(idleTimeMs + 1)
      TestUtils.waitUntilTrue(() => openChannel(request2).isEmpty, "Failed to close idle channel")
      TestUtils.waitUntilTrue(() => openOrClosingChannel(request2).nonEmpty, "Channel removed without processing staged receives")
      processRequest(overrideServer.requestChannel, request2) // this triggers a failed send since channel has been closed
      TestUtils.waitUntilTrue(() => openOrClosingChannel(request2).isEmpty, "Failed to remove channel with failed sends")
      assertNull("Received request after failed send", overrideServer.requestChannel.receiveRequest(200))

    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  @Test
  def testConnectionIdReuse() {
    val idleTimeMs = 60000
    val time = new MockTime()
    props.put(KafkaConfig.ConnectionsMaxIdleMsProp, idleTimeMs.toString)
    props.put("listeners", "PLAINTEXT://localhost:0")
    val serverMetrics = new Metrics
    @volatile var selector: TestableSelector = null
    val overrideConnectionId = "127.0.0.1:1-127.0.0.1:2-0"
    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, time, credentialProvider) {
      override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
                                protocol: SecurityProtocol, memoryPool: MemoryPool): Processor = {
        new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,
          config.connectionsMaxIdleMs, listenerName, protocol, config, metrics, credentialProvider, memoryPool, new LogContext()) {
          override protected[network] def connectionId(socket: Socket): String = overrideConnectionId
          override protected[network] def createSelector(channelBuilder: ChannelBuilder): Selector = {
           val testableSelector = new TestableSelector(config, channelBuilder, time, metrics)
           selector = testableSelector
           testableSelector
        }
        }
      }
    }

    def openChannel: Option[KafkaChannel] = overrideServer.processor(0).channel(overrideConnectionId)
    def openOrClosingChannel: Option[KafkaChannel] = overrideServer.processor(0).openOrClosingChannel(overrideConnectionId)
    def connectionCount = overrideServer.connectionCount(InetAddress.getByName("127.0.0.1"))

    // Create a client connection and wait for server to register the connection with the selector. For
    // test scenarios below where `Selector.register` fails, the wait ensures that checks are performed
    // only after `register` is processed by the server.
    def connectAndWaitForConnectionRegister(): Socket = {
      val connections = selector.operationCounts(SelectorOperation.Register)
      val socket = connect(overrideServer)
      TestUtils.waitUntilTrue(() =>
        selector.operationCounts(SelectorOperation.Register) == connections + 1, "Connection not registered")
      socket
    }

    try {
      overrideServer.startup()
      val socket1 = connectAndWaitForConnectionRegister()
      TestUtils.waitUntilTrue(() => connectionCount == 1 && openChannel.isDefined, "Failed to create channel")
      val channel1 = openChannel.getOrElse(throw new RuntimeException("Channel not found"))

      // Create new connection with same id when `channel1` is still open and in Selector.channels
      // Check that new connection is closed and openChannel still contains `channel1`
      connectAndWaitForConnectionRegister()
      TestUtils.waitUntilTrue(() => connectionCount == 1, "Failed to close channel")
      assertSame(channel1, openChannel.getOrElse(throw new RuntimeException("Channel not found")))

      // Send requests to `channel1` until a receive is staged and advance time beyond idle time so that `channel1` is
      // closed with staged receives and is in Selector.closingChannels
      val serializedBytes = producerRequestBytes()
      val request = sendRequestsUntilStagedReceive(overrideServer, socket1, serializedBytes)
      time.sleep(idleTimeMs + 1)
      TestUtils.waitUntilTrue(() => openChannel.isEmpty, "Idle channel not closed")
      TestUtils.waitUntilTrue(() => openOrClosingChannel.isDefined, "Channel removed without processing staged receives")

      // Create new connection with same id when `channel1` is in Selector.closingChannels
      // Check that new connection is closed and openOrClosingChannel still contains `channel1`
      connectAndWaitForConnectionRegister()
      TestUtils.waitUntilTrue(() => connectionCount == 1, "Failed to close channel")
      assertSame(channel1, openOrClosingChannel.getOrElse(throw new RuntimeException("Channel not found")))

      // Complete request with failed send so that `channel1` is removed from Selector.closingChannels
      processRequest(overrideServer.requestChannel, request)
      TestUtils.waitUntilTrue(() => connectionCount == 0 && openOrClosingChannel.isEmpty, "Failed to remove channel with failed send")

      // Check that new connections can be created with the same id since `channel1` is no longer in Selector
      connectAndWaitForConnectionRegister()
      TestUtils.waitUntilTrue(() => connectionCount == 1 && openChannel.isDefined, "Failed to open new channel")
      val newChannel = openChannel.getOrElse(throw new RuntimeException("Channel not found"))
      assertNotSame(channel1, newChannel)
      newChannel.disconnect()

    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  private def sendRequestsUntilStagedReceive(server: SocketServer, socket: Socket, requestBytes: Array[Byte]): RequestChannel.Request = {
    def sendTwoRequestsReceiveOne(): RequestChannel.Request = {
      sendRequest(socket, requestBytes, flush = false)
      sendRequest(socket, requestBytes, flush = true)
      receiveRequest(server.requestChannel)
    }
    val (request, hasStagedReceives) = TestUtils.computeUntilTrue(sendTwoRequestsReceiveOne()) { req =>
      val connectionId = req.context.connectionId
      val hasStagedReceives = server.processor(0).numStagedReceives(connectionId) > 0
      if (!hasStagedReceives) {
        processRequest(server.requestChannel, req)
        processRequest(server.requestChannel)
      }
      hasStagedReceives
    }
    assertTrue(s"Receives not staged for ${org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS} ms", hasStagedReceives)
    request
  }

  @Test
  def testSocketsCloseOnShutdown() {
    // open a connection
    val plainSocket = connect(protocol = SecurityProtocol.PLAINTEXT)
    plainSocket.setTcpNoDelay(true)
    val bytes = new Array[Byte](40)
    // send a request first to make sure the connection has been picked up by the socket server
    sendRequest(plainSocket, bytes, Some(0))
    processRequest(server.requestChannel)
    // the following sleep is necessary to reliably detect the connection close when we send data below
    Thread.sleep(200L)
    // make sure the sockets are open
    server.acceptors.asScala.values.foreach(acceptor => assertFalse(acceptor.serverChannel.socket.isClosed))
    // then shutdown the server
    shutdownServerAndMetrics(server)

    val largeChunkOfBytes = new Array[Byte](1000000)
    // doing a subsequent send should throw an exception as the connection should be closed.
    // send a large chunk of bytes to trigger a socket flush
    try {
      sendRequest(plainSocket, largeChunkOfBytes, Some(0))
      fail("expected exception when writing to closed plain socket")
    } catch {
      case _: IOException => // expected
    }
  }

  @Test
  def testMaxConnectionsPerIp() {
    // make the maximum allowable number of connections
    val conns = (0 until server.config.maxConnectionsPerIp).map(_ => connect())
    // now try one more (should fail)
    val conn = connect()
    conn.setSoTimeout(3000)
    assertEquals(-1, conn.getInputStream.read())
    conn.close()

    // it should succeed after closing one connection
    val address = conns.head.getInetAddress
    conns.head.close()
    TestUtils.waitUntilTrue(() => server.connectionCount(address) < conns.length,
      "Failed to decrement connection count after close")
    val conn2 = connect()
    val serializedBytes = producerRequestBytes()
    sendRequest(conn2, serializedBytes)
    val request = server.requestChannel.receiveRequest(2000)
    assertNotNull(request)
  }

  @Test
  def testZeroMaxConnectionsPerIp() {
    val newProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
    newProps.setProperty(KafkaConfig.MaxConnectionsPerIpProp, "0")
    newProps.setProperty(KafkaConfig.MaxConnectionsPerIpOverridesProp, "%s:%s".format("127.0.0.1", "5"))
    val server = new SocketServer(KafkaConfig.fromProps(newProps), new Metrics(), Time.SYSTEM, credentialProvider)
    try {
      server.startup()
      // make the maximum allowable number of connections
      val conns = (0 until 5).map(_ => connect(server))
      // now try one more (should fail)
      val conn = connect(server)
      conn.setSoTimeout(3000)
      assertEquals(-1, conn.getInputStream.read())
      conn.close()

      // it should succeed after closing one connection
      val address = conns.head.getInetAddress
      conns.head.close()
      TestUtils.waitUntilTrue(() => server.connectionCount(address) < conns.length,
        "Failed to decrement connection count after close")
      val conn2 = connect(server)
      val serializedBytes = producerRequestBytes()
      sendRequest(conn2, serializedBytes)
      val request = server.requestChannel.receiveRequest(2000)
      assertNotNull(request)

      // now try to connect from the external facing interface, which should fail
      val conn3 = connect(s = server, localAddr = InetAddress.getLocalHost)
      conn3.setSoTimeout(3000)
      assertEquals(-1, conn3.getInputStream.read())
      conn3.close()
    } finally {
      shutdownServerAndMetrics(server)
    }
  }

  @Test
  def testMaxConnectionsPerIpOverrides() {
    val overrideNum = server.config.maxConnectionsPerIp + 1
    val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
    overrideProps.put(KafkaConfig.MaxConnectionsPerIpOverridesProp, s"localhost:$overrideNum")
    val serverMetrics = new Metrics()
    val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM, credentialProvider)
    try {
      overrideServer.startup()
      // make the maximum allowable number of connections
      val conns = (0 until overrideNum).map(_ => connect(overrideServer))

      // it should succeed
      val serializedBytes = producerRequestBytes()
      sendRequest(conns.last, serializedBytes)
      val request = overrideServer.requestChannel.receiveRequest(2000)
      assertNotNull(request)

      // now try one more (should fail)
      val conn = connect(overrideServer)
      conn.setSoTimeout(3000)
      assertEquals(-1, conn.getInputStream.read())
    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  @Test
  def testSslSocketServer() {
    val trustStoreFile = File.createTempFile("truststore", ".jks")
    val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, interBrokerSecurityProtocol = Some(SecurityProtocol.SSL),
      trustStoreFile = Some(trustStoreFile))
    overrideProps.put(KafkaConfig.ListenersProp, "SSL://localhost:0")

    val serverMetrics = new Metrics
    val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM, credentialProvider)
    try {
      overrideServer.startup()
      val sslContext = SSLContext.getInstance("TLSv1.2")
      sslContext.init(null, Array(TestUtils.trustAllCerts), new java.security.SecureRandom())
      val socketFactory = sslContext.getSocketFactory
      val sslSocket = socketFactory.createSocket("localhost",
        overrideServer.boundPort(ListenerName.forSecurityProtocol(SecurityProtocol.SSL))).asInstanceOf[SSLSocket]
      sslSocket.setNeedClientAuth(false)

      val correlationId = -1
      val clientId = ""
      val ackTimeoutMs = 10000
      val ack = 0: Short
      val emptyRequest = ProduceRequest.Builder.forCurrentMagic(ack, ackTimeoutMs,
        new HashMap[TopicPartition, MemoryRecords]()).build()
      val emptyHeader = new RequestHeader(ApiKeys.PRODUCE, emptyRequest.version, clientId, correlationId)

      val byteBuffer = emptyRequest.serialize(emptyHeader)
      byteBuffer.rewind()
      val serializedBytes = new Array[Byte](byteBuffer.remaining)
      byteBuffer.get(serializedBytes)

      sendRequest(sslSocket, serializedBytes)
      processRequest(overrideServer.requestChannel)
      assertEquals(serializedBytes.toSeq, receiveResponse(sslSocket).toSeq)
      sslSocket.close()
    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  @Test
  def testSessionPrincipal() {
    val socket = connect()
    val bytes = new Array[Byte](40)
    sendRequest(socket, bytes, Some(0))
    assertEquals(KafkaPrincipal.ANONYMOUS, receiveRequest(server.requestChannel).session.principal)
  }

  /* Test that we update request metrics if the client closes the connection while the broker response is in flight. */
  @Test
  def testClientDisconnectionUpdatesRequestMetrics() {
    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
    val serverMetrics = new Metrics
    var conn: Socket = null
    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM, credentialProvider) {
      override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
                                protocol: SecurityProtocol, memoryPool: MemoryPool): Processor = {
        new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,
          config.connectionsMaxIdleMs, listenerName, protocol, config, metrics, credentialProvider, MemoryPool.NONE, new LogContext()) {
          override protected[network] def sendResponse(response: RequestChannel.Response, responseSend: Send) {
            conn.close()
            super.sendResponse(response, responseSend)
          }
        }
      }
    }
    try {
      overrideServer.startup()
      conn = connect(overrideServer)
      val serializedBytes = producerRequestBytes()
      sendRequest(conn, serializedBytes)

      val channel = overrideServer.requestChannel
      val request = receiveRequest(channel)

      val requestMetrics = channel.metrics(request.header.apiKey.name)
      def totalTimeHistCount(): Long = requestMetrics.totalTimeHist.count
      val expectedTotalTimeCount = totalTimeHistCount() + 1

      // send a large buffer to ensure that the broker detects the client disconnection while writing to the socket channel.
      // On Mac OS X, the initial write seems to always succeed and it is able to write up to 102400 bytes on the initial
      // write. If the buffer is smaller than this, the write is considered complete and the disconnection is not
      // detected. If the buffer is larger than 102400 bytes, a second write is attempted and it fails with an
      // IOException.
      val send = new NetworkSend(request.context.connectionId, ByteBuffer.allocate(550000))
      channel.sendResponse(new RequestChannel.Response(request, Some(send), SendAction, None))
      TestUtils.waitUntilTrue(() => totalTimeHistCount() == expectedTotalTimeCount,
        s"request metrics not updated, expected: $expectedTotalTimeCount, actual: ${totalTimeHistCount()}")

    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  @Test
  def testClientDisconnectionWithStagedReceivesFullyProcessed() {
    val serverMetrics = new Metrics
    @volatile var selector: TestableSelector = null
    val overrideConnectionId = "127.0.0.1:1-127.0.0.1:2-0"
    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM, credentialProvider) {
      override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
                                protocol: SecurityProtocol, memoryPool: MemoryPool): Processor = {
        new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,
          config.connectionsMaxIdleMs, listenerName, protocol, config, metrics, credentialProvider, memoryPool, new LogContext()) {
          override protected[network] def connectionId(socket: Socket): String = overrideConnectionId
          override protected[network] def createSelector(channelBuilder: ChannelBuilder): Selector = {
           val testableSelector = new TestableSelector(config, channelBuilder, time, metrics)
           selector = testableSelector
           testableSelector
        }
        }
      }
    }

    def openChannel: Option[KafkaChannel] = overrideServer.processor(0).channel(overrideConnectionId)
    def openOrClosingChannel: Option[KafkaChannel] = overrideServer.processor(0).openOrClosingChannel(overrideConnectionId)

    try {
      overrideServer.startup()
      val socket = connect(overrideServer)

      TestUtils.waitUntilTrue(() => openChannel.nonEmpty, "Channel not found")

      // Setup channel to client with staged receives so when client disconnects
      // it will be stored in Selector.closingChannels
      val serializedBytes = producerRequestBytes(1)
      val request = sendRequestsUntilStagedReceive(overrideServer, socket, serializedBytes)

      // Set SoLinger to 0 to force a hard disconnect via TCP RST
      socket.setSoLinger(true, 0)
      socket.close()

      // Complete request with socket exception so that the channel is removed from Selector.closingChannels
      processRequest(overrideServer.requestChannel, request)
      TestUtils.waitUntilTrue(() => openOrClosingChannel.isEmpty, "Channel not closed after failed send")
      assertTrue("Unexpected completed send", selector.completedSends.isEmpty)
    } finally {
      overrideServer.shutdown()
      serverMetrics.close()
    }
  }

  /*
   * Test that we update request metrics if the channel has been removed from the selector when the broker calls
   * `selector.send` (selector closes old connections, for example).
   */
  @Test
  def testBrokerSendAfterChannelClosedUpdatesRequestMetrics() {
    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
    props.setProperty(KafkaConfig.ConnectionsMaxIdleMsProp, "100")
    val serverMetrics = new Metrics
    var conn: Socket = null
    val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM, credentialProvider)
    try {
      overrideServer.startup()
      conn = connect(overrideServer)
      val serializedBytes = producerRequestBytes()
      sendRequest(conn, serializedBytes)
      val channel = overrideServer.requestChannel
      val request = receiveRequest(channel)

      TestUtils.waitUntilTrue(() => overrideServer.processor(request.processor).channel(request.context.connectionId).isEmpty,
        s"Idle connection `${request.context.connectionId}` was not closed by selector")

      val requestMetrics = channel.metrics(request.header.apiKey.name)
      def totalTimeHistCount(): Long = requestMetrics.totalTimeHist.count
      val expectedTotalTimeCount = totalTimeHistCount() + 1

      processRequest(channel, request)

      TestUtils.waitUntilTrue(() => totalTimeHistCount() == expectedTotalTimeCount,
        s"request metrics not updated, expected: $expectedTotalTimeCount, actual: ${totalTimeHistCount()}")

    } finally {
      shutdownServerAndMetrics(overrideServer)
    }
  }

  @Test
  def testRequestMetricsAfterStop(): Unit = {
    server.stopProcessingRequests()
    val version = ApiKeys.PRODUCE.latestVersion
    val version2 = (version - 1).toShort
    for (_ <- 0 to 1) server.requestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version).mark()
    server.requestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version2).mark()
    assertEquals(2, server.requestChannel.metrics(ApiKeys.PRODUCE.name).requestRate(version).count())
    server.requestChannel.updateErrorMetrics(ApiKeys.PRODUCE, Map(Errors.NONE -> 1))
    val nonZeroMeters = Map(s"kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce,version=$version" -> 2,
        s"kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce,version=$version2" -> 1,
        "kafka.network:type=RequestMetrics,name=ErrorsPerSec,request=Produce,error=NONE" -> 1)

    def requestMetricMeters = YammerMetrics
      .defaultRegistry
      .allMetrics.asScala
      .filterKeys(k => k.getType == "RequestMetrics")
      .collect { case (k, metric: Meter) => (k.toString, metric.count) }

    assertEquals(nonZeroMeters, requestMetricMeters.filter { case (_, value) => value != 0 })
    server.shutdown()
    assertEquals(Map.empty, requestMetricMeters)
  }

  @Test
  def testMetricCollectionAfterShutdown(): Unit = {
    server.shutdown()

    val nonZeroMetricNamesAndValues = YammerMetrics
      .defaultRegistry
      .allMetrics.asScala
      .filterKeys(k => k.getName.endsWith("IdlePercent") || k.getName.endsWith("NetworkProcessorAvgIdlePercent"))
      .collect { case (k, metric: Gauge[_]) => (k, metric.value().asInstanceOf[Double]) }
      .filter { case (_, value) => value != 0.0 }

    assertEquals(Map.empty, nonZeroMetricNamesAndValues)
  }

  @Test
  def testProcessorMetricsTags(): Unit = {
    val kafkaMetricNames = metrics.metrics.keySet.asScala.filter(_.tags.asScala.get("listener").nonEmpty)
    assertFalse(kafkaMetricNames.isEmpty)

    val expectedListeners = Set("PLAINTEXT", "TRACE")
    kafkaMetricNames.foreach { kafkaMetricName =>
      assertTrue(expectedListeners.contains(kafkaMetricName.tags.get("listener")))
    }

    // legacy metrics not tagged
    val yammerMetricsNames = YammerMetrics.defaultRegistry.allMetrics.asScala
      .filterKeys(_.getType.equals("Processor"))
      .collect { case (k, _: Gauge[_]) => k }
    assertFalse(yammerMetricsNames.isEmpty)

    yammerMetricsNames.foreach { yammerMetricName =>
      assertFalse(yammerMetricName.getMBeanName.contains("listener="))
    }
  }

  /**
   * Tests exception handling in [[Processor.configureNewConnections]]. Exception is
   * injected into [[Selector.register]] which is used to register each new connection.
   * Test creates two connections in a single iteration by waking up the selector only
   * when two connections are ready.
   * Verifies that
   * - first failed connection is closed
   * - second connection is processed successfully after the first fails with an exception
   * - processor is healthy after the exception
   */
  @Test
  def configureNewConnectionException(): Unit = {
    withTestableServer { testableServer =>
      val testableSelector = testableServer.testableSelector

      testableSelector.updateMinWakeup(2)
      testableSelector.addFailure(SelectorOperation.Register)
      val sockets = (1 to 2).map(_ => connect(testableServer))
      testableSelector.waitForOperations(SelectorOperation.Register, 2)
      TestUtils.waitUntilTrue(() => testableServer.connectionCount(localAddress) == 1, "Failed channel not removed")

      assertProcessorHealthy(testableServer, testableSelector.notFailed(sockets))
    }
  }

  /**
   * Tests exception handling in [[Processor.processNewResponses]]. Exception is
   * injected into [[Selector.send]] which is used to send the new response.
   * Test creates two responses in a single iteration by waking up the selector only
   * when two responses are ready.
   * Verifies that
   * - first failed channel is closed
   * - second response is processed successfully after the first fails with an exception
   * - processor is healthy after the exception
   */
  @Test
  def processNewResponseException(): Unit = {
    withTestableServer { testableServer =>
      val testableSelector = testableServer.testableSelector
      testableSelector.updateMinWakeup(2)

      val sockets = (1 to 2).map(_ => connect(testableServer))
      sockets.foreach(sendRequest(_, producerRequestBytes()))

      testableServer.testableSelector.addFailure(SelectorOperation.Send)
      sockets.foreach(_ => processRequest(testableServer.requestChannel))
      testableSelector.waitForOperations(SelectorOperation.Send, 2)
      testableServer.waitForChannelClose(testableSelector.allFailedChannels.head, locallyClosed = true)

      assertProcessorHealthy(testableServer, testableSelector.notFailed(sockets))
    }
  }

  /**
   * Tests exception handling in [[Processor.processNewResponses]] when [[Selector.send]]
   * fails with `CancelledKeyException`, which is handled by the selector using a different
   * code path. Test scenario is similar to [[SocketServerTest.processNewResponseException]].
   */
  @Test
  def sendCancelledKeyException(): Unit = {
    withTestableServer { testableServer =>
      val testableSelector = testableServer.testableSelector
      testableSelector.updateMinWakeup(2)

      val sockets = (1 to 2).map(_ => connect(testableServer))
      sockets.foreach(sendRequest(_, producerRequestBytes()))
      val requestChannel = testableServer.requestChannel

      val requests = sockets.map(_ => receiveRequest(requestChannel))
      val failedConnectionId = requests(0).context.connectionId
      // `KafkaChannel.disconnect()` cancels the selection key, triggering CancelledKeyException during send
      testableSelector.channel(failedConnectionId).disconnect()
      requests.foreach(processRequest(requestChannel, _))
      testableSelector.waitForOperations(SelectorOperation.Send, 2)
      testableServer.waitForChannelClose(failedConnectionId, locallyClosed = false)

      val successfulSocket = if (isSocketConnectionId(failedConnectionId, sockets(0))) sockets(1) else sockets(0)
      assertProcessorHealthy(testableServer, Seq(successfulSocket))
    }
  }

  /**
   * Tests exception handling in [[Processor.processNewResponses]] when [[Selector.send]]
   * to a channel in closing state throws an exception. Test scenario is similar to
   * [[SocketServerTest.processNewResponseException]].
   */
  @Test
  def closingChannelException(): Unit = {
    withTestableServer { testableServer =>
      val testableSelector = testableServer.testableSelector
      testableSelector.updateMinWakeup(2)

      val sockets = (1 to 2).map(_ => connect(testableServer))
      val serializedBytes = producerRequestBytes()
      val request = sendRequestsUntilStagedReceive(testableServer, sockets(0), serializedBytes)
      sendRequest(sockets(1), serializedBytes)

      testableSelector.addFailure(SelectorOperation.Send)
      sockets(0).close()
      processRequest(testableServer.requestChannel, request)
      processRequest(testableServer.requestChannel) // Also process request from other channel
      testableSelector.waitForOperations(SelectorOperation.Send, 2)
      testableServer.waitForChannelClose(request.context.connectionId, locallyClosed = true)

      assertProcessorHealthy(testableServer, Seq(sockets(1)))
    }
  }

  /**
   * Tests exception handling in [[Processor.processCompletedReceives]]. Exception is
   * injected into [[Selector.mute]] which is used to mute the channel when a receive is complete.
   * Test creates two receives in a single iteration by caching completed receives until two receives
   * are complete.
   * Verifies that
   * - first failed channel is closed
   * - second receive is processed successfully after the first fails with an exception
   * - processor is healthy after the exception
   */
  @Test
  def processCompletedReceiveException(): Unit = {
    withTestableServer { testableServer =>
      val sockets = (1 to 2).map(_ => connect(testableServer))
      val testableSelector = testableServer.testableSelector
      val requestChannel = testableServer.requestChannel

      testableSelector.cachedCompletedReceives.minPerPoll = 2
      testableSelector.addFailure(SelectorOperation.Mute)
      sockets.foreach(sendRequest(_, producerRequestBytes()))
      val requests = sockets.map(_ => receiveRequest(requestChannel))
      testableSelector.waitForOperations(SelectorOperation.Mute, 2)
      testableServer.waitForChannelClose(testableSelector.allFailedChannels.head, locallyClosed = true)
      requests.foreach(processRequest(requestChannel, _))

      assertProcessorHealthy(testableServer, testableSelector.notFailed(sockets))
    }
  }

  /**
   * Tests exception handling in [[Processor.processCompletedSends]]. Exception is
   * injected into [[Selector.unmute]] which is used to unmute the channel after send is complete.
   * Test creates two completed sends in a single iteration by caching completed sends until two
   * sends are complete.
   * Verifies that
   * - first failed channel is closed
   * - second send is processed successfully after the first fails with an exception
   * - processor is healthy after the exception
   */
  @Test
  def processCompletedSendException(): Unit = {
    withTestableServer { testableServer =>
      val testableSelector = testableServer.testableSelector
      val sockets = (1 to 2).map(_ => connect(testableServer))
      val requests = sockets.map(sendAndReceiveRequest(_, testableServer))

      testableSelector.addFailure(SelectorOperation.Unmute)
      requests.foreach(processRequest(testableServer.requestChannel, _))
      testableSelector.waitForOperations(SelectorOperation.Unmute, 2)
      testableServer.waitForChannelClose(testableSelector.allFailedChannels.head, locallyClosed = true)

      assertProcessorHealthy(testableServer, testableSelector.notFailed(sockets))
    }
  }

  /**
   * Tests exception handling in [[Processor.processDisconnected]]. An invalid connectionId
   * is inserted to the disconnected list just before the actual valid one.
   * Verifies that
   * - first invalid connectionId is ignored
   * - second disconnected channel is processed successfully after the first fails with an exception
   * - processor is healthy after the exception
   */
  @Test
  def processDisconnectedException(): Unit = {
    withTestableServer { testableServer =>
      val (socket, connectionId) = connectAndProcessRequest(testableServer)
      val testableSelector = testableServer.testableSelector

      // Add an invalid connectionId to `Selector.disconnected` list before the actual disconnected channel
      // and check that the actual id is processed and the invalid one ignored.
      testableSelector.cachedDisconnected.minPerPoll = 2
      testableSelector.cachedDisconnected.deferredValues += "notAValidConnectionId" -> ChannelState.EXPIRED
      socket.close()
      testableSelector.operationCounts.clear()
      testableSelector.waitForOperations(SelectorOperation.Poll, 1)
      testableServer.waitForChannelClose(connectionId, locallyClosed = false)

      assertProcessorHealthy(testableServer)
    }
  }

  /**
   * Tests that `Processor` continues to function correctly after a failed [[Selector.poll]].
   */
  @Test
  def pollException(): Unit = {
    withTestableServer { testableServer =>
      val (socket, _) = connectAndProcessRequest(testableServer)
      val testableSelector = testableServer.testableSelector

      testableSelector.addFailure(SelectorOperation.Poll)
      testableSelector.operationCounts.clear()
      testableSelector.waitForOperations(SelectorOperation.Poll, 2)

      assertProcessorHealthy(testableServer, Seq(socket))
    }
  }

  /**
   * Tests handling of `ControlThrowable`. Verifies that the selector is closed.
   */
  @Test
  def controlThrowable(): Unit = {
    withTestableServer { testableServer =>
      connectAndProcessRequest(testableServer)
      val testableSelector = testableServer.testableSelector

      testableSelector.operationCounts.clear()
      testableSelector.addFailure(SelectorOperation.Poll,
          Some(new RuntimeException("ControlThrowable exception during poll()") with ControlThrowable))
      testableSelector.waitForOperations(SelectorOperation.Poll, 1)

      testableSelector.waitForOperations(SelectorOperation.CloseSelector, 1)
    }
  }

  private def withTestableServer(testWithServer: TestableSocketServer => Unit): Unit = {
    props.put("listeners", "PLAINTEXT://localhost:0")
    val testableServer = new TestableSocketServer
    testableServer.startup()
    try {
        testWithServer(testableServer)
    } finally {
      shutdownServerAndMetrics(testableServer)
    }
  }

  private def assertProcessorHealthy(testableServer: TestableSocketServer, healthySockets: Seq[Socket] = Seq.empty): Unit = {
    val selector = testableServer.testableSelector
    selector.reset()
    val requestChannel = testableServer.requestChannel

    // Check that existing channels behave as expected
    healthySockets.foreach { socket =>
      val request = sendAndReceiveRequest(socket, testableServer)
      processRequest(requestChannel, request)
      socket.close()
    }
    TestUtils.waitUntilTrue(() => testableServer.connectionCount(localAddress) == 0, "Channels not removed")

    // Check new channel behaves as expected
    val (socket, connectionId) = connectAndProcessRequest(testableServer)
    assertArrayEquals(producerRequestBytes(), receiveResponse(socket))
    assertNotNull("Channel should not have been closed", selector.channel(connectionId))
    assertNull("Channel should not be closing", selector.closingChannel(connectionId))
    socket.close()
    TestUtils.waitUntilTrue(() => testableServer.connectionCount(localAddress) == 0, "Channels not removed")
  }

  // Since all sockets use the same local host, it is sufficient to check the local port
  def isSocketConnectionId(connectionId: String, socket: Socket): Boolean =
    connectionId.contains(s":${socket.getLocalPort}-")

  class TestableSocketServer extends SocketServer(KafkaConfig.fromProps(props),
      new Metrics, Time.SYSTEM, credentialProvider) {

    @volatile var selector: Option[TestableSelector] = None

    override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, listenerName: ListenerName,
                                protocol: SecurityProtocol, memoryPool: MemoryPool): Processor = {
      new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,
        config.connectionsMaxIdleMs, listenerName, protocol, config, metrics, credentialProvider, memoryPool, new
            LogContext()) {

        override protected[network] def createSelector(channelBuilder: ChannelBuilder): Selector = {
           val testableSelector = new TestableSelector(config, channelBuilder, time, metrics)
           assertEquals(None, selector)
           selector = Some(testableSelector)
           testableSelector
        }
      }
    }

    def testableSelector: TestableSelector =
      selector.getOrElse(throw new IllegalStateException("Selector not created"))

    def waitForChannelClose(connectionId: String, locallyClosed: Boolean): Unit = {
      val selector = testableSelector
      if (locallyClosed) {
        TestUtils.waitUntilTrue(() => selector.allLocallyClosedChannels.contains(connectionId),
            s"Channel not closed: $connectionId")
        assertTrue("Unexpected disconnect notification", testableSelector.allDisconnectedChannels.isEmpty)
      } else {
        TestUtils.waitUntilTrue(() => selector.allDisconnectedChannels.contains(connectionId),
            s"Disconnect notification not received: $connectionId")
        assertTrue("Channel closed locally", testableSelector.allLocallyClosedChannels.isEmpty)
      }
      val openCount = selector.allChannels.size - 1 // minus one for the channel just closed above
      TestUtils.waitUntilTrue(() => connectionCount(localAddress) == openCount, "Connection count not decremented")
      TestUtils.waitUntilTrue(() =>
        processor(0).inflightResponseCount == 0, "Inflight responses not cleared")
      assertNull("Channel not removed", selector.channel(connectionId))
      assertNull("Closing channel not removed", selector.closingChannel(connectionId))
    }
  }

  sealed trait SelectorOperation
  object SelectorOperation {
    case object Register extends SelectorOperation
    case object Poll extends SelectorOperation
    case object Send extends SelectorOperation
    case object Mute extends SelectorOperation
    case object Unmute extends SelectorOperation
    case object Wakeup extends SelectorOperation
    case object Close extends SelectorOperation
    case object CloseSelector extends SelectorOperation
  }

  class TestableSelector(config: KafkaConfig, channelBuilder: ChannelBuilder, time: Time, metrics: Metrics)
        extends Selector(config.socketRequestMaxBytes, config.connectionsMaxIdleMs,
            metrics, time, "socket-server", new HashMap, false, true, channelBuilder, MemoryPool.NONE, new LogContext()) {

    val failures = mutable.Map[SelectorOperation, Exception]()
    val operationCounts = mutable.Map[SelectorOperation, Int]().withDefaultValue(0)
    val allChannels = mutable.Set[String]()
    val allLocallyClosedChannels = mutable.Set[String]()
    val allDisconnectedChannels = mutable.Set[String]()
    val allFailedChannels = mutable.Set[String]()

    // Enable data from `Selector.poll()` to be deferred to a subsequent poll() until
    // the number of elements of that type reaches `minPerPoll`. This enables tests to verify
    // that failed processing doesn't impact subsequent processing within the same iteration.
    class PollData[T] {
      var minPerPoll = 1
      val deferredValues = mutable.Buffer[T]()
      val currentPollValues = mutable.Buffer[T]()
      def update(newValues: mutable.Buffer[T]): Unit = {
        if (currentPollValues.nonEmpty || deferredValues.size + newValues.size >= minPerPoll) {
          if (deferredValues.nonEmpty) {
            currentPollValues ++= deferredValues
            deferredValues.clear()
          }
          currentPollValues ++= newValues
        } else
          deferredValues ++= newValues
      }
      def reset(): Unit = {
        currentPollValues.clear()
      }
    }
    val cachedCompletedReceives = new PollData[NetworkReceive]()
    val cachedCompletedSends = new PollData[Send]()
    val cachedDisconnected = new PollData[(String, ChannelState)]()
    val allCachedPollData = Seq(cachedCompletedReceives, cachedCompletedSends, cachedDisconnected)
    @volatile var minWakeupCount = 0
    @volatile var pollTimeoutOverride: Option[Long] = None

    def addFailure(operation: SelectorOperation, exception: Option[Exception] = None) {
      failures += operation ->
        exception.getOrElse(new IllegalStateException(s"Test exception during $operation"))
    }

    private def onOperation(operation: SelectorOperation, connectionId: Option[String], onFailure: => Unit): Unit = {
      operationCounts(operation) += 1
      failures.remove(operation).foreach { e =>
        connectionId.foreach(allFailedChannels.add)
        onFailure
        throw e
      }
    }

    def waitForOperations(operation: SelectorOperation, minExpectedTotal: Int): Unit = {
      TestUtils.waitUntilTrue(() =>
        operationCounts.getOrElse(operation, 0) >= minExpectedTotal, "Operations not performed within timeout")
    }

    def runOp[T](operation: SelectorOperation, connectionId: Option[String],
        onFailure: => Unit = {})(code: => T): T = {
      // If a failure is set on `operation`, throw that exception even if `code` fails
      try code
      finally onOperation(operation, connectionId, onFailure)
    }

    override def register(id: String, socketChannel: SocketChannel): Unit = {
      runOp(SelectorOperation.Register, Some(id), onFailure = close(id)) {
        super.register(id, socketChannel)
      }
    }

    override def send(s: Send): Unit = {
      runOp(SelectorOperation.Send, Some(s.destination)) {
        super.send(s)
      }
    }

    override def poll(timeout: Long): Unit = {
      try {
        allCachedPollData.foreach(_.reset)
        runOp(SelectorOperation.Poll, None) {
          super.poll(pollTimeoutOverride.getOrElse(timeout))
        }
      } finally {
        super.channels.asScala.foreach(allChannels += _.id)
        allDisconnectedChannels ++= super.disconnected.asScala.keys
        cachedCompletedReceives.update(super.completedReceives.asScala)
        cachedCompletedSends.update(super.completedSends.asScala)
        cachedDisconnected.update(super.disconnected.asScala.toBuffer)
      }
    }

    override def mute(id: String): Unit = {
      runOp(SelectorOperation.Mute, Some(id)) {
        super.mute(id)
      }
    }

    override def unmute(id: String): Unit = {
      runOp(SelectorOperation.Unmute, Some(id)) {
        super.unmute(id)
      }
    }

    override def wakeup(): Unit = {
      runOp(SelectorOperation.Wakeup, None) {
        if (minWakeupCount > 0)
          minWakeupCount -= 1
        if (minWakeupCount <= 0)
          super.wakeup()
      }
    }

    override def disconnected: java.util.Map[String, ChannelState] = cachedDisconnected.currentPollValues.toMap.asJava

    override def completedSends: java.util.List[Send] = cachedCompletedSends.currentPollValues.asJava

    override def completedReceives: java.util.List[NetworkReceive] = cachedCompletedReceives.currentPollValues.asJava

    override def close(id: String): Unit = {
      runOp(SelectorOperation.Close, Some(id)) {
        super.close(id)
        allLocallyClosedChannels += id
      }
    }

    override def close(): Unit = {
      runOp(SelectorOperation.CloseSelector, None) {
        super.close()
      }
    }

    def updateMinWakeup(count: Int): Unit = {
      minWakeupCount = count
      // For tests that ignore wakeup to process responses together, increase poll timeout
      // to ensure that poll doesn't complete before the responses are ready
      pollTimeoutOverride = Some(1000L)
      // Wakeup current poll to force new poll timeout to take effect
      super.wakeup()
    }

    def reset(): Unit = {
      failures.clear()
      allCachedPollData.foreach(_.minPerPoll = 1)
    }

    def notFailed(sockets: Seq[Socket]): Seq[Socket] = {
      // Each test generates failure for exactly one failed channel
      assertEquals(1, allFailedChannels.size)
      val failedConnectionId = allFailedChannels.head
      sockets.filterNot(socket => isSocketConnectionId(failedConnectionId, socket))
    }
  }
}


text:
import re
import debug, tests
from common import TestCommon, TimeoutError
from results import RowResults
from results import RawResults

@tests.add_test
class PhaseChangeTest(TestCommon):
    ''' Scheduler phase change latency benchmark '''
    name = "phasechange"

    def get_build_targets(self, build, machine):
        targets = super(PhaseChangeTest, self).get_build_targets(build, machine)
        targets.append("%s/sbin/phases_bench" % (machine.get_bootarch()))
        return targets

    def get_finish_string(self):
        return "client done."

    def run(self, build, machine, testdir):
        ncores = machine.get_ncores()
        for i in [2**x for x in range(0,ncores + 1) if 2 ** x <= ncores]:
            debug.log('running %s on %d/%d cores' % (self.name, i, ncores))
            modules = self.get_modules(build, machine)
            modules.add_module("phases_bench", [i])
            self.boot(machine, modules)
            for line in self.collect_data(machine):
                yield line

    def process_data(self, testdir, rawiter):
        results = RawResults('threads')
        data = []
        for line in rawiter:
            m = re.match("duration \d+: (\d+)", line)
            if m:
                data.append(int(m.group(1)))
                continue

            m = re.match("number of threads: (\d+)", line)
            if m:
                results.add_group(m.group(1), data)
                data = []

        return results

# Run every benchmark for 10 seconds (specified in ms)
PHASESCALE_TIMEOUT = 10000

@tests.add_test
class PhaseScaleTest(TestCommon):
    ''' Scheduler phase change scalability benchmark '''
    name = "phasechange_scale"

    def get_build_targets(self, build, machine):
        targets = super(PhaseScaleTest, self).get_build_targets(build, machine)
        targets.append("%s/sbin/phases_scale_bench" % (machine.get_bootarch()))
        return targets

    def get_finish_string(self):
        return "client done."

    def run(self, build, machine, testdir):
        ncores = machine.get_ncores()
        for delay in [0, 1, 5, 10, 25, 50, 100, 250, 500]:
            for i in range(2, ncores + 1):
                debug.log('running %s on %d/%d cores, delay %d' % (self.name, i, ncores, delay))
                modules = self.get_modules(build, machine)
                modules.add_module("phases_scale_bench", [i, delay, PHASESCALE_TIMEOUT])
                self.boot(machine, modules)
                for line in self.collect_data(machine):
                    yield line

    def process_data(self, testdir, rawiter):
        results = RowResults(['threads', 'delay', 'slowdown'])
        process_sum = 0
        sums = []
        baseline = []

        for line in rawiter:
            m = re.match("workcnt (\d+): (\d+)", line)
            if m:
                if int(m.group(1)) != 0:
                    process_sum += int(m.group(2))
                continue

            m = re.match("number of threads: (\d+), delay: (\d+)", line)
            if m:
                sums.append([m.group(1), m.group(2), process_sum])
                if int(m.group(2)) == 0:
                    baseline.append([m.group(1), process_sum])
                process_sum = 0

        for sum in sums:
            for [t,p] in baseline:
                if t == sum[0]:
                    basesum = p
                    break

            procsum = float(sum[2]) / float(basesum)
            results.add_row([sum[0], sum[1], procsum])

        return results

redpajama_set_name: RedPajamaGithub


text:
/**
 *
 * @param setupObject = {
   *  pie: {
   *    mainIconSize: 40,
   *    mainIcon: "\uf21c",
   *    radiusOuterPx: 100,
   *    radiusInnerPx: 50,
   *    mainIconDX: -25,
   *    mainIconDY: -20
   *  }
   * }
 * @param dataObject
 */
function visualCuePieLikeElement( setupObject, dataArrObject ){

  if( setupObject[ "x" ] === undefined ){
    setupObject[ "x" ] = 0;
  }
  if( setupObject[ "y" ] === undefined ){
    setupObject[ "y" ] = 0;
  }
  if( setupObject[ "element" ] === undefined ){
    setupObject[ "element" ] = "body";
  }
  if( setupObject[ "toolTipClass" ] === undefined ){
    setupObject[ "toolTipClass" ] = "tooltip";
  }
  if( setupObject[ "angleStartDeg" ] === undefined ){
    setupObject[ "angleStartDeg" ] = 0;
  }
  if( setupObject[ "angleEndDeg" ] === undefined ){
    setupObject[ "angleEndDeg" ] = 360;
  }
  if( setupObject[ "toolTipEffectInDuration" ] === undefined ){
    setupObject[ "toolTipEffectInDuration" ] = 100;
  }
  if( setupObject[ "toolTipEffectOutDuration" ] === undefined ){
    setupObject[ "toolTipEffectOutDuration" ] = 500;
  }
  if( setupObject[ "toolTipOpacity" ] === undefined ){
    setupObject[ "toolTipOpacity" ] = .8;
  }
  if( setupObject[ "toolTipDX" ] === undefined ){
    setupObject[ "toolTipDX" ] = 0;
  }
  if( setupObject[ "toolTipDY" ] === undefined ){
    setupObject[ "toolTipDY" ] = 0;
  }
  if( setupObject[ "toolTipFillColor" ] === undefined ){
    setupObject[ "toolTipFillColor" ] = "#888888";
  }
  /*if( setupObject[ "mainIconSize" ] === undefined ){
   setupObject[ "mainIconSize" ] = 80;
   }*/
  if( setupObject[ "mainIconColor" ] === undefined ){
    setupObject[ "mainIconColor" ] = "#000000";
  }
  if( setupObject[ "mainIconDX" ] === undefined ){
    setupObject[ "mainIconDX" ] = 0;
  }
  if( setupObject[ "mainIconDY" ] === undefined ){
    setupObject[ "mainIconDY" ] = 0;
  }
  /*if( setupObject[ "mainIcon" ] === undefined ){
   setupObject[ "mainIcon" ] = "\uf21c";
   }*/

  var donut = d3.pie()
  .startAngle( setupObject[ "angleStartDeg" ] * ( Math.PI / 180 ) )
  .endAngle( setupObject[ "angleEndDeg" ] * ( Math.PI / 180 ) );

  var arc = d3.arc()
  .innerRadius( setupObject[ "radiusInnerPx" ] )
  .outerRadius( setupObject[ "radiusOuterPx" ] );

  var div = d3.select( setupObject[ "element" ] ).append( "div" )
  .attr( "class", setupObject[ "toolTipClass" ] )
  .style( "opacity", 0 );

  var vis = d3.select( setupObject[ "element" ] )
  .append( "svg:svg" )
  .data( [ dataArrObject ] )
  .attr( "width", setupObject[ "radiusOuterPx" ] * 2 )
  .attr( "height", setupObject[ "radiusOuterPx" ] * 2 );

  var arcs = vis.selectAll( "g.arc" )
  .data( donut.value(
    function( d ) {
      return d.percentual;
    }
  ))
  .enter().append( "svg:g" )
  .attr( "class", "arc" )
  .attr( "transform", "translate(" + ( setupObject[ "radiusOuterPx" ] + setupObject[ "x" ] ) + "," + ( setupObject[ "radiusOuterPx" ] + setupObject[ "y" ] ) + ")" )
  .on( "mousemove", function( d ) {
    div.transition()
    .duration( setupObject[ "toolTipEffectInDuration" ] )
    .style( "opacity", setupObject[ "toolTipOpacity" ] );

    div.html( d.data.status )
    .style( "left", ( d3.event.pageX + setupObject[ "toolTipDX" ] ) + "px" )
    .style( "top", ( d3.event.pageY + setupObject[ "toolTipDY" ] ) + "px" );
  })
  .on( "mouseout", function( d ) {
    div.transition()
    .duration( setupObject[ "toolTipEffectOutDuration" ] )
    .style( "opacity", 0 );
  });

  arcs.append( "svg:path" )
  .attr( "fill", function( d, i ) {
    return d.data.background;
  })
  .attr( "d", arc );

  arcs.append( "svg:foreignObject" )
  .attr( "font-family", "FontAwesome" )
  .attr( 'font-size', function( d ) {
    return ( d.data.size ) + 'px'
  })
  .attr( 'color', function( d ) {
    return d.data.color
  })
  .attr( "transform", function( d ) {
    var textWidth = getTextWidth( d.data.icon, "FontAwesome" );
    return "translate(" + ( arc.centroid( d )[ 0 ] - textWidth / 2 + d.data.dx ) + "," + ( arc.centroid( d )[ 1 ] - textWidth / 2 + d.data.dy ) + ")";
  })
  .attr( "text-anchor", function( d ) {
    return ( d.endAngle + d.startAngle ) / 2 > Math.PI ? "end" : "start";
  })
  .text( function( d ) {
    return d.data.icon;
  })
  .attr( 'pointer-events', 'none' );

  vis.append( "circle" )
  .attr( "cx", function( d ) {
    return setupObject[ "radiusOuterPx" ] + setupObject[ "x" ];
  })
  .attr( "cy", function( d ) {
    return setupObject[ "radiusOuterPx" ] + setupObject[ "y" ];
  })
  .attr( "r", setupObject[ "radiusInnerPx" ] )
  .style( "fill", setupObject[ "toolTipFillColor" ] )
  .on( "mousemove", function( d ) {
    div.transition()
    .duration( setupObject[ "toolTipEffectInDuration" ] )
    .style( "opacity", setupObject[ "toolTipOpacity" ] );
    div.html( "Olá mundo!" )
    .style("left", ( d3.event.pageX + setupObject[ "toolTipDX" ] ) + "px")
    .style("top", ( d3.event.pageY + setupObject[ "toolTipDY" ] ) + "px");
  })
  .on( "mouseout", function( d ) {
    div.transition()
    .duration( setupObject[ "toolTipEffectOutDuration" ] )
    .style( "opacity", 0 );
  });
  vis.append( "svg:foreignObject" )
  .attr( "font-family", "FontAwesome" )
  .attr( 'font-size', function( d ) {
    return ( setupObject[ "mainIconSize" ] ) + 'px'
  })
  .attr( 'pointer-events', 'none' )
  .attr( 'color', function( d ) {
    return setupObject[ "mainIconColor" ];
  })
  .attr("x", function( d ) {
    return ( setupObject[ "radiusOuterPx" ] + setupObject[ "mainIconDX" ] + setupObject[ "x" ] );
  })
  .attr("y", function( d ) {
    return ( setupObject[ "radiusOuterPx" ] + setupObject[ "mainIconDY" ] + setupObject[ "y" ] );
  })
  .text( setupObject[ "mainIcon" ] );

  return {
    vis: vis,
    html: div
  };
}


text:
<?xml version="1.0" encoding="utf-8"?>
<GridLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:orientation="vertical"
    android:layout_width="fill_parent"
    android:layout_height="fill_parent"
    android:useDefaultMargins="true"
    android:alignmentMode="alignBounds"
    android:columnOrderPreserved="false"
    android:columnCount="4"
    >
    <TextView
        android:text="Type"
        android:layout_gravity="right"/>
    <EditText
        android:id="@+id/type"
        />
    <TextView
        android:text="Running water"
        android:layout_column="0"
        android:layout_gravity="right"
        />
    <EditText
        android:id="@+id/water"
        />
</GridLayout>


text:
<?xml version="1.0" ?>
<tei>
	<teiHeader>
		<fileDesc xml:id="55000631"/>
	</teiHeader>
	<text xml:lang="en">
		<front>
<lb/>
	<docTitle>
	<titlePart>High Precision Narrow-Band RF Ranging<lb/></titlePart>
	</docTitle>

	<byline>
	<docAuthor>Bradley D. Farnsworth, </docAuthor>
	</byline>

	<byline>
	<affiliation>ENSCO, Inc.<lb/></affiliation>
	</byline>

	<byline>
	<docAuthor>David W.A. Taylor, </docAuthor>
	</byline>

	<byline>
	<affiliation>ENSCO, Inc.<lb/></affiliation>
	</byline>

	<note type="other">BIOGRAPHY<lb/>Bradley D. Farnsworth</note>

	<email>(farnsworth.bradley@ensco.com)<lb/></email>

	<note type = "other">is an ENSCO Senior Engineer focusing on high-precision<lb/> RF ranging and timing. He was the Sensors Technical<lb/> Lead for the national semifinalist Team Case for the 2007<lb/> DARPA Urban Challenge and won a Best Student Paper<lb/> award at the 2008 IEEE Sensors Conference. Mr.<lb/> Farnsworth holds the B.S. in Electrical Engineering<lb/> summa cum laude from Case Western Reserve University<lb/> and anticipates graduating with the M.S. in Electrical<lb/> Engineering from Case in May 2010.<lb/> Dr. David W.A. Taylor </note>

	<email>(taylor.david@ensco.com)</email>

	<note type="other"> is<lb/> ENSCO's Director of Technology Development, where<lb/> he leads the company's R&D programs developing<lb/> sensors and systems for national security applications.<lb/> Dr. Taylor holds three U.S. patents and is an expert in<lb/> GPS-denied navigation technologies, particularly as<lb/> applied to military and intelligence applications. Dr.<lb/> Taylor holds the B.S. in Physics from Rhodes College and<lb/> the Ph.D. in Geophysics from Virginia Tech.<lb/> </note>

	<div type="abstract"> ABSTRACT<lb/> A round-trip, time-of-flight, high-precision, ranging radio<lb/> operating in the 915 MHz ISM band is presented. A<lb/> custom digital baseband processor is implemented on a<lb/> Xilinx Virtex-4 FPGA and interfaced with COTS RF<lb/> hardware. Processing gain enabled by direct-sequence<lb/> spread-spectrum (DSSS) signaling directly improves the<lb/> ranging accuracy. The system is designed with free-<lb/>running baseband clocks that do not require fine<lb/> synchronization.<lb/> Prototype brassboard system<lb/> measurement accuracies of 7 cm (1-σ) have been obtained<lb/> during testing in an open field.<lb/>
	</div>

		</front>
	</text>
</tei>


text:
(function ($, window) {
	window.MyBB = window.MyBB || {};

	window.MyBB.Cookie = {
		cookiePrefix: '',
		cookiePath: '/',
		cookieDomain: '',

		init: function () {
			MyBB.Settings = MyBB.Settings || {};
			if (typeof MyBB.Settings.cookiePrefix != 'undefined') {
				this.cookiePrefix = MyBB.Settings.cookiePrefix;
			}
			if (typeof MyBB.Settings.cookiePath != 'undefined') {
				this.cookiePath = MyBB.Settings.cookiePath;
			}
			if (typeof MyBB.Settings.cookieDomain != 'undefined') {
				this.cookieDomain = MyBB.Settings.cookieDomain;
			}
		},

		get: function (name) {
			this.init();

			name = this.cookiePrefix + name;
			return $.cookie(name);
		},

		set: function (name, value, expires) {
			this.init();

			name = this.cookiePrefix + name;
			if (!expires) {
				expires = 157680000; // 5*365*24*60*60 => 5 years
			}

			expire = new Date();
			expire.setTime(expire.getTime() + (expires * 1000));

			options = {
				expires: expire,
				path: this.cookiePath,
				domain: this.cookieDomain
			};

			return $.cookie(name, value, options);
		},

		unset: function (name) {
			this.init();

			name = this.cookiePrefix + name;

			options = {
				path: this.cookiePath,
				domain: this.cookieDomain
			};
			return $.removeCookie(name, options);
		}
	}
})
(jQuery, window);
(function ($, window) {
	window.MyBB = window.MyBB || {};

	window.MyBB.Spinner = {
		inProgresses: 0,
		add: function () {
			this.inProgresses++;
			if (this.inProgresses == 1) {
				$("#spinner").show();
			}
		},
		remove: function () {
			this.inProgresses--;
			if (this.inProgresses == 0) {
				$("#spinner").hide();
			}
		}
	}
})
(jQuery, window);
(function($, window) {
    window.MyBB = window.MyBB || {};

	window.MyBB.Modals = function Modals()
	{
		$("*[data-modal]").on("click", this.toggleModal).bind(this);
		$.modal.defaults.closeText = 'x';
	};

	window.MyBB.Modals.prototype.toggleModal = function toggleModal(event) {
		event.preventDefault();

		// Check to make sure we're clicking the link and not a child of the link
		if(event.target.nodeName === "A")
		{
			// Woohoo, it's the link!
			var modalOpener = event.target,
				modalSelector = $(modalOpener).data("modal"),
				modalFind = $(modalOpener).data("modal-find"),
				modal = $('<div/>', {
	    			"class": "modal-dialog",
					closeText: ''
				}),
				modalContent = "";
		} else {
			// Nope, it's one of those darn children.
			var modalOpener = event.target,
				modalSelector = $(modalOpener).parent().data("modal"),
				modalFind = $(modalOpener).data("modal-find"),
				modal = $('<div/>', {
	    			"class": "modal-dialog",
					closeText: ''
				}),
				modalContent = "";
		}

		if (modalSelector.substring(0, 1) === "." || modalSelector.substring(0, 1) === "#") {
			// Assume using a local, existing HTML element.
			modalContent = $(modalSelector).html();
			modal.html(modalContent);
			modal.appendTo("body").modal({
				zIndex: 1000,
				closeText: ''
			});
			$('.modalHide').hide();
			$("input[type=number]").stepper();
			$(".password-toggle").hideShowPassword(false, true);
			new window.MyBB.Avatar();
		} else {
			// Assume modal content is coming from an AJAX request

			// data-modal-find is optional, default to "#content"
			if (modalFind === undefined) {
				modalFind = "#content";
			}

			MyBB.Spinner.add();

			var modalParams = $(event.currentTarget).attr('data-modal-params');
			if (modalParams) {
				modalParams = JSON.parse(modalParams);
				console.log(modalParams);
			} else {
				modalParams = {};
			}

			$.get('/'+modalSelector, modalParams, function(response) {
				var responseObject = $(response);

				modalContent = $(modalFind, responseObject).html();
				modal.html(modalContent);
				modal.appendTo("body").modal({
					zIndex: 1000,
					closeText: ''
				});
				$('.modalHide').hide();
				$("input[type=number]").stepper();
				$(".password-toggle").hideShowPassword(false, true);
				new window.MyBB.Avatar();

				// Remove modal after close
				modal.on($.modal.CLOSE, function() {
					$(this).remove();
				})
			}).always(function() {
				MyBB.Spinner.remove();
			});
		}

	};

    var modals = new window.MyBB.Modals(); // TODO: put this elsewhere :)
})(jQuery, window);

(function($, window) {
    window.MyBB = window.MyBB || {};

	window.MyBB.Posts = function Posts()
	{
		// Show and hide posts
		$(".postToggle").on("click", this.togglePost).bind(this);

		// Confirm Delete
		$(".delete a").on("click", $.proxy(this.confirmDelete, this));
	};

	// Show and hide posts
	window.MyBB.Posts.prototype.togglePost = function togglePost(event) {
		event.preventDefault();
		// Are we minimized or not?
		if($(event.target).hasClass("fa-minus"))
		{
			// Perhaps instead of hide, apply a CSS class?
			$(event.target).parent().parent().parent().addClass("post--hidden");
			// Make button a plus sign for expanding
			$(event.target).addClass("fa-plus");
			$(event.target).removeClass("fa-minus");

		} else {
			// We like this person again
			$(event.target).parent().parent().parent().removeClass("post--hidden");
			// Just in case we change our mind again, show the hide button
			$(event.target).addClass("fa-minus");
			$(event.target).removeClass("fa-show");
		}
	};

	// Confirm Delete
	window.MyBB.Posts.prototype.confirmDelete = function confirmDelete(event) {
		return confirm(Lang.get('topic.confirmDelete'));
	};

	var posts = new window.MyBB.Posts();

})(jQuery, window);
(function($, window) {
    window.MyBB = window.MyBB || {};

	window.MyBB.Polls = function Polls()
	{
		this.optionElement = $('#option-simple').clone().attr('id', '').removeClass('hidden').addClass('poll-option').hide();
		$('#option-simple').remove();

		this.removeOption($('#add-poll .poll-option'));

		$('#new-option').click($.proxy(this.addOption, this));

		$('#poll-maximum-options').hide();

		$('#poll-is-multiple').change($.proxy(this.toggleMaxOptionsInput, this)).change();

		var $addPollButton = $("#add-poll-button");
		$addPollButton.click($.proxy(this.toggleAddPoll, this));
		if($addPollButton.length) {
			if($('#add-poll-input').val() === '1') {
				$('#add-poll').slideDown();
			}
		}

		this.timePicker();
	};

	window.MyBB.Polls.prototype.timePicker = function timePicker() {
		$('#poll-end-at').datetimepicker({
			format: 'Y-m-d H:i:s',
			lang: 'mybb',
			i18n: {
				mybb: {
					months: [
						Lang.get('general.months.january'),
						Lang.get('general.months.february'),
						Lang.get('general.months.march'),
						Lang.get('general.months.april'),
						Lang.get('general.months.may'),
						Lang.get('general.months.june'),
						Lang.get('general.months.july'),
						Lang.get('general.months.august'),
						Lang.get('general.months.september'),
						Lang.get('general.months.october'),
						Lang.get('general.months.november'),
						Lang.get('general.months.december')
					],
					dayOfWeek: [
						Lang.get('general.dayOfWeek.sun'),
						Lang.get('general.dayOfWeek.mon'),
						Lang.get('general.dayOfWeek.tue'),
						Lang.get('general.dayOfWeek.wed'),
						Lang.get('general.dayOfWeek.thu'),
						Lang.get('general.dayOfWeek.fri'),
						Lang.get('general.dayOfWeek.sat')
					]
				}
			},
			minDate: 0
		});
	};

	window.MyBB.Polls.prototype.toggleAddPoll = function toggleAddPoll() {
		if($('#add-poll-input').val() === '1') {
			$('#add-poll-input').val(0);
			$('#add-poll').slideUp();
		} else {
			$('#add-poll-input').val(1);
			$('#add-poll').slideDown();
		}
		return false;
	};

	window.MyBB.Polls.prototype.addOption = function addOption(event) {
		var num_options = $('#add-poll .poll-option').length;
		if(num_options >= 10) { // TODO: settings
			alert(Lang.choice('poll.errorManyOptions', 10)); // TODO: JS Error
			return false;
		}
		var $option = this.optionElement.clone();
		$option.find('input').attr('name', 'option['+(num_options+1)+']');
		$('#add-poll .poll-option').last().after($option);
		$option.slideDown();
		this.removeOption($option);
		return false;
	};

	window.MyBB.Polls.prototype.removeOption = function bindRemoveOption($parent) {
		$parent.find('.remove-option').click($.proxy(function(event) {
			var $me = $(event.target),
				$myParent = $me.parents('.poll-option');
			if($('.poll-option').length <= 2) // TODO: settings
			{
				alert(Lang.choice('poll.errorFewOptions', 2)); // TODO: JS Error
				return false;
			}

			$myParent.slideUp(500);

			setTimeout($.proxy(function() {
				$myParent.remove();
				this.fixOptionsName();
			}, this), 500);
		}, this));
		if(!Modernizr.touch) {
			$parent.find('.remove-option').powerTip({ placement: 's', smartPlacement: true });
		}
	};

	window.MyBB.Polls.prototype.fixOptionsName = function() {
		var i = 0;
		$('#add-poll .poll-option').each(function() {
			i++;
			$(this).find('input').attr('name', 'option['+i+']');
		});
	};

	window.MyBB.Polls.prototype.toggleMaxOptionsInput = function toggleMaxOptionsInput(event) {
		me = event.target;
		if($(me).is(':checked')) {
			$('#poll-maximum-options').slideDown();
		}
		else {
			$('#poll-maximum-options').slideUp();
		}
	};

	var polls = new window.MyBB.Polls();

})(jQuery, window);
(function ($, window) {
	window.MyBB = window.MyBB || {};

	window.MyBB.Quotes = function Quotes() {

		// MultiQuote
		$(".quote-button").on("click", this.multiQuoteButton.bind(this));

		this.showQuoteBar();

		$(".quote-bar__select").on("click", $.proxy(this.addQuotes, this));
		$(".quote-bar__view").on("click", $.proxy(this.viewQuotes, this));
		$(".quote-bar__deselect").on("click", $.proxy(this.removeQuotes, this));

		$('.quick-quote .fast').on('click', $.proxy(this.quickQuote, this));
		$('.quick-quote .add').on('click', $.proxy(this.quickAddQuote, this));

		$('.quote__select').on("click", $.proxy(this.quoteAdd, this));
		$('.quote__remove').on("click", $.proxy(this.quoteRemove, this));
		$("body").on("mouseup", $.proxy(this.checkQuickQuote, this));

		this.quoteButtons();
	};

	window.MyBB.Quotes.prototype.quickQuote = function quickQuote(event) {
		var $me = $(event.target);
		if (!$me.hasClass('quick-quote')) {
			$me = $me.parents('.quick-quote');
		}

		var $post = $me.parents('.post');

		if ($me.data('content')) {
			$content = $('<div/>');
			$content.html($me.data('content'));
			this.addQuote($post.data('postid'), $post.data('type'), $content.text());
		}
		this.hideQuickQuote();
	}

	window.MyBB.Quotes.prototype.quickAddQuote = function quickAddQuote(event) {
		var $me = $(event.target),
			quotes = this.getQuotes();
		if (!$me.hasClass('quick-quote')) {
			$me = $me.parents('.quick-quote');
		}

		var $post = $me.parents('.post');

		if ($me.data('content')) {
			$content = $('<div/>');
			$content.html($me.data('content'));
			quotes.push({
				'id': $post.data('type') + '_' + $post.data('postid'),
				'data': $content.text()
			});
			MyBB.Cookie.set('quotes', JSON.stringify(quotes));

			this.showQuoteBar();
		}
		this.hideQuickQuote();
	}

	window.MyBB.Quotes.prototype.checkQuickQuote = function checkQuickQuote(event) {
		var $me = $(event.target);
		if ($me.hasClass('quick-quote') || $me.parents('.quick-quote').length) {
			return false;
		}
		if (!$me.hasClass('post')) {
			$me = $me.parents('.post');
		}

		if ($me && $me.length) {
			var pid = $me.data('postid');

			if ($.trim(window.getSelection().toString())) {
				if (elementContainsSelection($me.find('.post__body')[0])) {
					this.showQuickQuote(pid);
				}
				else {
					this.hideQuickQuote();
				}
			}
			else {
				this.hideQuickQuote();
			}
		}
		else {
			this.hideQuickQuote();
		}
	}

	window.MyBB.Quotes.prototype.showQuickQuote = function showQuckQuote(pid) {
		var selection = window.getSelection(),
			range = selection.getRangeAt(0),
			rect = range.getBoundingClientRect();
		$elm = $("#post-" + pid).find('.quick-quote').show().data('content', $.trim(window.getSelection().toString()));
		$elm.css({
			'top': (window.scrollY + rect.top - $elm.outerHeight() - 4) + 'px',
			'left': (window.scrollX + rect.left - (($elm.outerWidth() - rect.width) / 2)) + 'px'
		});
	}

	window.MyBB.Quotes.prototype.hideQuickQuote = function () {
		$('.post .quick-quote').hide().data('content', '');
	}

	window.MyBB.Quotes.prototype.getQuotes = function getQuotes() {
		var quotes = MyBB.Cookie.get('quotes'),
			myQuotes = [];
		if (!quotes) {
			quotes = [];
		}
		else {
			quotes = JSON.parse(quotes);
		}
		$.each(quotes, function (key, quote) {
			if (quote != null) {
				myQuotes.push(quote);
			}
		});

		MyBB.Cookie.set('quotes', JSON.stringify(myQuotes));
		return myQuotes;
	};

	// MultiQuote
	window.MyBB.Quotes.prototype.multiQuoteButton = function multiQuoteButton(event) {
		event.preventDefault();
		var $me = $(event.target);
		if (!$me.hasClass('quote-button')) {
			$me = $me.parents('.quote-button');
		}
		var $post = $me.parents('.post');

		var postId = parseInt($post.data('postid')),
			type = $post.data('type'),
			quotes = this.getQuotes();

		if (postId) {
			var removed = false;
			$.each(quotes, function(key, quote) {
				if(typeof quote != 'string') {
					return;
				}
				if(quote == type + '_' + postId) {
					delete quotes[key];
					removed = true;
				}
			});
			if (!removed) {
				quotes.push(type + '_' + postId);
				$me.find('.quote-button__add').hide();
				$me.find('.quote-button__remove').show();
			}
			else {
				$me.find('.quote-button__add').show();
				$me.find('.quote-button__remove').hide();
			}

			MyBB.Cookie.set('quotes', JSON.stringify(quotes));

			this.showQuoteBar();
			return false;
		}

	};

	window.MyBB.Quotes.prototype.showQuoteBar = function showQuoteBar() {
		var quotes = this.getQuotes();

		if (quotes.length) {
			$(".quote-bar").show();
		}
		else {
			$(".quote-bar").hide();
		}
	};

	window.MyBB.Quotes.prototype.addQuotes = function addQuotes() {
		var quotes = this.getQuotes(),
			$quoteBar = $(".quote-bar"),
			$textarea = $($quoteBar.data('textarea'));

		MyBB.Spinner.add();

		$.ajax({
			url: '/post/quotes',
			data: {
				'posts': quotes,
				'_token': $quoteBar.parents('form').find('input[name=_token]').val()
			},
			method: 'POST'
		}).done(function (json) {
			if (json.error) {
				alert(json.error);// TODO: js error
			}
			else {
				var value = $textarea.val();
				if (value && value.substr(-2) != "\n\n") {
					if (value.substr(-1) != "\n") {
						value += "\n";
					}
					value += "\n";
				}
				$textarea.val(value + json.message).focus();
			}
			$.modal.close();
		}).always(function () {
			MyBB.Spinner.remove();
		});

		$quoteBar.hide();
		MyBB.Cookie.unset('quotes');
		this.quoteButtons();
		return false;
	};

	window.MyBB.Quotes.prototype.addQuote = function addQuote(postid, type, content) {
		var $textarea = $("#message");

		MyBB.Spinner.add();

		$.ajax({
			url: '/post/quotes',
			data: {
				'posts': [
					{
						'id': type + '_' + postid,
						'data': content
					}
				],
				'_token': $(".quote-bar").parents('form').find('input[name=_token]').val()
			},
			method: 'POST'
		}).done(function (json) {
			if (json.error) {
				alert(json.error);// TODO: js error
			}
			else {
				var value = $textarea.val();
				if (value && value.substr(-2) != "\n\n") {
					if (value.substr(-1) != "\n") {
						value += "\n";
					}
					value += "\n";
				}
				$textarea.val(value + json.message).focus();
			}
		}).always(function () {
			MyBB.Spinner.remove();
		});

		this.hideQuickQuote();

		return false;
	};

	window.MyBB.Quotes.prototype.viewQuotes = function viewQuotes() {
		MyBB.Spinner.add();

		$.ajax({
			url: '/post/quotes/all',
			data: {
				'posts': this.getQuotes(),
				'_token': $(".quote-bar").parents('form').find('input[name=_token]').val()
			},
			method: 'POST'
		}).done($.proxy(function (data) {
			var modalContent = $("#content", $(data)),
				modal = $('<div/>', {
					"class": "modal-dialog view-quotes",
					closeText: ''
				});
			modalContent.find('.view-quotes__quotes').css({
				'max-height': ($(window).height()-250)+'px'
			});
			modal.html(modalContent.html());
			modal.appendTo("body").modal({
				zIndex: 1000,
				closeText: ''
			});

			if(Modernizr.touch)
			{
				$('.radio-buttons .radio-button, .checkbox-buttons .checkbox-button').click(function() {

				});
			}
			else
			{
				$('span.icons i, a, .caption, time').powerTip({ placement: 's', smartPlacement: true });
			}

			$('.quote__select').on("click", $.proxy(this.quoteAdd, this));
			$('.quote__remove').on("click", $.proxy(this.quoteRemove, this));
			$(".select-all-quotes").on("click", $.proxy(this.addQuotes, this));
			$('.modal-hide').hide();
		}, this)).always(function () {
			MyBB.Spinner.remove();
		});

		this.hideQuickQuote();

		return false;
	};

	window.MyBB.Quotes.prototype.removeQuotes = function removeQuotes() {
		$quoteBar = $(".quote-bar");
		$quoteBar.hide();
		MyBB.Cookie.unset('quotes');
		this.quoteButtons();
		return false;
	};

	window.MyBB.Quotes.prototype.quoteButtons = function quoteButtons() {
		var quotes = this.getQuotes();

		$('.quote-button__add').show();
		$('.quote-button__remove').hide();

		$.each(quotes, function (key, quote) {
			if (typeof quote != 'string') {
				return;
			}
			quote = quote.split('_');
			type = quote[0];
			postId = parseInt(quote[1]);
			var $quoteButton = $("#post-" + postId + "[data-type='" + type + "']").find('.quoteButton');
			$quoteButton.find('.quote-button__add').hide();
			$quoteButton.find('.quote-button__remove').show();
		})
	}

	window.MyBB.Quotes.prototype.quoteAdd = function quoteAdd(event) {
		var $me = $(event.target),
			$post = $me.parents('.content-quote'),
			$textarea = $("#message"),
			quotes = this.getQuotes();

		var value = $textarea.val();
		if (value && value.substr(-2) != "\n\n") {
			if (value.substr(-1) != "\n") {
				value += "\n";
			}
			value += "\n";
		}
		$textarea.val(value + $post.data('quote')).focus();

		delete quotes[$post.data('id')];
		MyBB.Cookie.set('quotes', JSON.stringify(quotes));
		$post.slideUp('fast');

		if(this.getQuotes().length == 0) {
			$.modal.close();
		}

		while($post.next().length) {
			$post = $post.next();
			$post.data('id', $post.data('id')-1);
		}

		this.quoteButtons();
		this.showQuoteBar();
	}

	window.MyBB.Quotes.prototype.quoteRemove = function quoteRemove(event) {
		var $me = $(event.target),
			$post = $me.parents('.content-quote'),
			quotes = this.getQuotes();

		delete quotes[$post.data('id')];
		MyBB.Cookie.set('quotes', JSON.stringify(quotes));
		$post.slideUp('fast');

		if(this.getQuotes().length == 0) {
			$.modal.close();
		}

		while($post.next().length) {
			$post = $post.next();
			$post.data('id', $post.data('id')-1);
		}

		this.quoteButtons();
		this.showQuoteBar();
		return false;
	}

	var quotes = new window.MyBB.Quotes();

	// Helper functions
	// http://stackoverflow.com/questions/8339857
	function isOrContains(node, container) {
		while (node) {
			if (node === container) {
				return true;
			}
			node = node.parentNode;
		}
		return false;
	}

	function elementContainsSelection(el) {
		var sel;
		if (window.getSelection) {
			sel = window.getSelection();
			if (sel.rangeCount > 0) {
				for (var i = 0; i < sel.rangeCount; ++i) {
					if (!isOrContains(sel.getRangeAt(i).commonAncestorContainer, el)) {
						return false;
					}
				}
				return true;
			}
		} else if ((sel = document.selection) && sel.type != "Control") {
			return isOrContains(sel.createRange().parentElement(), el);
		}
		return false;
	}

})
(jQuery, window);
(function ($, window) {
	window.MyBB = window.MyBB || {};

	window.MyBB.Avatar = function Avatar() {
		this.dropAvatar();
	};

	window.MyBB.Avatar.prototype.dropAvatar = function dropAvatar()
	{
		var $avatarDrop = $('#avatar-drop');
		if ($avatarDrop.length == 0 || $avatarDrop.is(':visible') == $('#avatar-drop-area').is(':visible')) {
			return;
		}

		var avatarDropUrl = $avatarDrop.attr('action');
		Dropzone.autoDiscover = false;
		var avatarDrop = new Dropzone("#avatar-drop", {
			url: avatarDropUrl + "?ajax=1",
			acceptedFiles: "image/*",
			clickable: '#file-input',
			paramName: "avatar_file",
			thumbnailWidth: null,
			thumbnailHeight: null,
			uploadMultiple: false,
			init: function () {
				$("#avatar-drop-area").show();
			},
			previewTemplate: '<div style="display:none" />'
		});

		avatarDrop.on("thumbnail", function (file, dataUrl) {
			$("#crop").find('.jcrop').find('img').attr('src', dataUrl);//TODO: use better selection
		});

		avatarDrop.on("sending", function (file) {
			MyBB.Spinner.add();
		});
		avatarDrop.on("complete", function () {
			MyBB.Spinner.remove();
		});
		avatarDrop.on("success", function (file) {
			var data = $.parseJSON(file.xhr.responseText);
			if (data.needCrop == true) {
				$("<a />").data('modal', '#crop').click((new MyBB.Modals()).toggleModal).click(); // TODO: create a method in modal.js for it
				var $jcrop = $('.modal').find('.jcrop');

				function JcropUpdateInputs(c) {
					$jcrop.find('.jcrop-x').val(c.x);
					$jcrop.find('.jcrop-y').val(c.y);
					$jcrop.find('.jcrop-x2').val(c.x2);
					$jcrop.find('.jcrop-y2').val(c.y2);
					$jcrop.find('.jcrop-w').val(c.w);
					$jcrop.find('.jcrop-h').val(c.h);
				}

				$jcrop.find('img').Jcrop({
					onChange: JcropUpdateInputs,
					onSelect: JcropUpdateInputs
				});

				$('.modal').find('.crop-img').click(function () {
					MyBB.Spinner.add();
					$.post('/account/avatar/crop?ajax=1', {
						x: $jcrop.find('.jcrop-x').val(),
						y: $jcrop.find('.jcrop-y').val(),
						x2: $jcrop.find('.jcrop-x2').val(),
						y2: $jcrop.find('.jcrop-y2').val(),
						w: $jcrop.find('.jcrop-w').val(),
						h: $jcrop.find('.jcrop-h').val(),
						"_token": $('input[name="_token"]').val()
					}).done(function (data) {
						if (data.success) {
							alert(data.message); // TODO: JS Message
							$.modal.close();
							$(".my-avatar").attr('src', data.avatar);
						} else {
							alert(data.error);// TODO: JS Error
						}
					}).always(function () {
						MyBB.Spinner.remove();
					});
				});
			}
		});
	}

	var avatar = new window.MyBB.Avatar();

})(jQuery, window);
$('html').addClass('js');

$(function () {

	$('.nojs').hide();

	if(Modernizr.touch)
	{
		$('.radio-buttons .radio-button, .checkbox-buttons .checkbox-button').click(function() {

		});
	}
	else
	{
		$('span.icons i, a, .caption, time').powerTip({ placement: 's', smartPlacement: true });
	}

	$('.user-navigation__links, #main-menu').dropit({ submenuEl: 'div.dropdown' });
	$('.dropdown-menu').dropit({ submenuEl: 'ul.dropdown', triggerEl: 'span.dropdown-button' });
	$("input[type=number]").stepper();
	$(".password-toggle").hideShowPassword(false, true);

	$("#search .search-button").click(function(event) {
		event.preventDefault();
		$("#search .search-container").slideDown();
	});

	autosize($('.post textarea'));

	$('a.show-menu__link').click(function(e) {
		if(menu == 0)
		{
			menu = 1;
			openMenu(e);
		}

		else
		{
			menu = 0;
			closeMenu(e);
		}
	});

/*	$('.post.reply textarea.editor, .form textarea.editor').sceditor({
		plugins: 'bbcode',
		style: 'js/vendor/sceditor/jquery.sceditor.default.min.css',
		emoticonsRoot: 'assets/images/',
		toolbar: 'bold,italic,underline|font,size,color,removeformat|left,center,right|image,link,unlink|emoticon,youtube|bulletlist,orderedlist|quote,code|source',
		resizeWidth: false,
		autofocus: false,
		autofocusEnd: false
	});*/
});

// Overwrite the powertip helper function - it's nearly the same
function getTooltipContent(element) {
	var tipText = element.data(DATA_POWERTIP),
		tipObject = element.data(DATA_POWERTIPJQ),
		tipTarget = element.data(DATA_POWERTIPTARGET),
		targetElement,
		content;

	if (tipText) {
		if ($.isFunction(tipText)) {
			tipText = tipText.call(element[0]);
		}
		content = tipText;
	} else if (tipObject) {
		if ($.isFunction(tipObject)) {
			tipObject = tipObject.call(element[0]);
		}
		if (tipObject.length > 0) {
			content = tipObject.clone(true, true);
		}
	} else if (tipTarget) {
		targetElement = $('#' + tipTarget);
		if (targetElement.length > 0) {
			content = targetElement.html();
		}
	}

	// Except we're escaping html
	return escapeHTML(content);
}

// Source: http://stackoverflow.com/questions/24816/escaping-html-strings-with-jquery

var entityMap = {
	"&": "&",
	"<": "<",
	">": ">",
	'"': '"',
	"'": ''',
	"/": '/'
};

function escapeHTML(string) {
	if(typeof string == 'string') {
		return String(string).replace(/[&<>"'\/]/g, function (s) {
			return entityMap[s];
		});
	}

	return string;
}

function submitFormAsGet(id, newRoute) {
	var form = $('#' + id);
	form.find("input[name=_token]").val('');

	if(newRoute != null) {
		form.attr('action', newRoute);
	}

	form.attr('method', 'get').submit();
	return false;
}

function openMenu(e) {
	e.preventDefault();
	$("body").animate({'background-position-x': '0px'}, 200, function() { });
	$(".sidebar-menu").animate({marginLeft: "0px"}, 200, function() { });
	$(".page-body").animate({marginLeft: "225px", marginRight: "-225px"}, 200, function() { });
}

function closeMenu(e) {
	e.preventDefault();
	$("body").animate({'background-position-x': '-225px'}, 200, function() { });
	$(".sidebar-menu").animate({marginLeft: "-225px"}, 200, function() { });
	$(".page-body").animate({marginLeft: "0", marginRight: "0"}, 200, function() { });
}

(function($, window) {
    window.MyBB = window.MyBB || {};

    window.MyBB.Moderation = function Moderation()
    {
        $.ajaxSetup({
            headers: {
                'X-CSRF-TOKEN': $('meta[name="csrf-token"]').attr('content')
            }
        });

        // inline moderation click handling
        $('a[data-moderate]').click($.proxy(function (e) {
            e.preventDefault();

            MyBB.Spinner.add();

            var moderation_content = $('[data-moderation-content]').first();
            $.post('/moderate', {
                moderation_name: $(e.currentTarget).attr('data-moderate'),
                moderation_content: moderation_content.attr('data-moderation-content'),
                moderation_ids: window.MyBB.Moderation.getSelectedIds(),
                moderation_source_type: moderation_content.attr('data-moderation-source-type'),
                moderation_source_id: moderation_content.attr('data-moderation-source-id')
            }, function (response) {
                document.location.reload();
            });
        }, this));

        // inline reverse moderation click handling
        $('a[data-moderate-reverse]').click($.proxy(function (e) {
            e.preventDefault();

            MyBB.Spinner.add();

            var moderation_content = $('[data-moderation-content]').first();
            $.post('/moderate/reverse', {
                moderation_name: $(e.currentTarget).attr('data-moderate-reverse'),
                moderation_content: moderation_content.attr('data-moderation-content'),
                moderation_ids: window.MyBB.Moderation.getSelectedIds(),
                moderation_source_type: moderation_content.attr('data-moderation-source-type'),
                moderation_source_id: moderation_content.attr('data-moderation-source-id')
            }, function (response) {
                document.location.reload();
            });
        }, this));

        // moderation bar clear selection handling
        $('.clear-selection a').click(function(e) {
            $('[data-moderation-content] input[type=checkbox]:checked').removeAttr('checked');
            $('[data-moderation-content] .highlight').removeClass('highlight');
            $('.inline-moderation').removeClass('floating');
        });

        // post level inline moderation checkbox handling
        $(".post :checkbox").change(function() {
            $(this).closest(".post").toggleClass("highlight", this.checked);

            var checked_boxes = $('.highlight').length;

            if(checked_boxes == 1)
            {
                $('.inline-moderation').addClass('floating');
            }

            if (checked_boxes > 1)
            {
                $('li[data-moderation-multi]').show();
            } else {
                $('li[data-moderation-multi]').hide();
            }

            if(checked_boxes == 0)
            {
                $('.inline-moderation').removeClass('floating');
            }

            $('.inline-moderation .selection-count').text(' ('+checked_boxes+')')
        });

        // topic level inline moderation checkbox handling
        $(".topic-list .topic :checkbox").change(function() {
            $(this).closest(".topic").toggleClass("highlight", this.checked);

            var checked_boxes = $('.highlight').length;

            if(checked_boxes == 1)
            {
                $('.inline-moderation').addClass('floating');
            }

            if (checked_boxes > 1)
            {
                $('li[data-moderation-multi]').show();
            } else {
                $('li[data-moderation-multi]').hide();
            }

            if(checked_boxes == 0)
            {
                $('.inline-moderation').removeClass('floating');
            }

            $('.inline-moderation .selection-count').text(' ('+checked_boxes+')')
        });

        $('li[data-moderation-multi]').hide();
    };

    // get the IDs of elements currently selected
    window.MyBB.Moderation.getSelectedIds = function getSelectedIds()
    {
        return $('input[type=checkbox][data-moderation-id]:checked').map(function () {
            return $(this).attr('data-moderation-id');
        }).get();
    };

    // grab the current selection and inject it into the modal so we can submit through a normal form
    window.MyBB.Moderation.injectModalParams = function injectFormData(element)
    {
        var moderation_content = $('[data-moderation-content]').first();
        $(element).attr('data-modal-params', JSON.stringify({
            moderation_content: moderation_content.attr('data-moderation-content'),
            moderation_ids: window.MyBB.Moderation.getSelectedIds(),
            moderation_source_type: moderation_content.attr('data-moderation-source-type'),
            moderation_source_id: moderation_content.attr('data-moderation-source-id')
        }));
    };

    var moderation = new window.MyBB.Moderation();

})(jQuery, window);