<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.activemq.apollo.openwire

import OpenwireConstants._

import org.fusesource.hawtdispatch._
import org.fusesource.hawtbuf._
import collection.mutable.{ListBuffer, HashMap}

import org.apache.activemq.apollo.broker._
import BufferConversions._
import java.io.IOException
import org.apache.activemq.apollo.selector.SelectorParser
import org.apache.activemq.apollo.filter.{BooleanExpression, FilterException}
import org.apache.activemq.apollo.transport._
import org.apache.activemq.apollo.broker.store._
import org.apache.activemq.apollo.util._
import java.util.concurrent.TimeUnit
import java.util.Map.Entry
import protocol._
import scala.util.continuations._
import security.SecurityContext
import tcp.TcpTransport
import codec.OpenWireFormat
import command._
import org.apache.activemq.apollo.openwire.dto.OpenwireConnectionStatusDTO
import org.apache.activemq.apollo.dto.{TopicDestinationDTO, DurableSubscriptionDestinationDTO, DestinationDTO}

object OpenwireProtocolHandler extends Log {

  val DEFAULT_DIE_DELAY = 5 * 1000L
  var die_delay = DEFAULT_DIE_DELAY

  val preferred_wireformat_settings = new WireFormatInfo();
  preferred_wireformat_settings.setVersion(OpenWireFormat.DEFAULT_VERSION);
  preferred_wireformat_settings.setStackTraceEnabled(true);
  preferred_wireformat_settings.setCacheEnabled(true);
  preferred_wireformat_settings.setTcpNoDelayEnabled(true);
  preferred_wireformat_settings.setTightEncodingEnabled(true);
  preferred_wireformat_settings.setSizePrefixDisabled(false);
  preferred_wireformat_settings.setMaxInactivityDuration(30 * 1000 * 1000);
  preferred_wireformat_settings.setMaxInactivityDurationInitalDelay(10 * 1000 * 1000);
  preferred_wireformat_settings.setCacheSize(1024);

}


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

class OpenwireProtocolHandler extends ProtocolHandler {

  var minimum_protocol_version = 1

  import OpenwireProtocolHandler._

  def dispatchQueue: DispatchQueue = connection.dispatch_queue

  def protocol = PROTOCOL

  var outbound_sessions: SinkMux[Command] = null
  var connection_session: Sink[Command] = null
  var closed = false

  var last_command_id=0

  def next_command_id = {
    last_command_id += 1
    last_command_id
  }

  var producerRoutes = new LRUCache[List[DestinationDTO], DeliveryProducerRoute](10) {
    override def onCacheEviction(eldest: Entry[List[DestinationDTO], DeliveryProducerRoute]) = {
      host.router.disconnect(eldest.getKey.toArray, eldest.getValue)
    }
  }

  var host: VirtualHost = null

  private def queue = connection.dispatch_queue

  var session_id: AsciiBuffer = _
  var wire_format: OpenWireFormat = _
  var login: Option[AsciiBuffer] = None
  var passcode: Option[AsciiBuffer] = None
  var dead = false
  val security_context = new SecurityContext

  var heart_beat_monitor: HeartBeatMonitor = new HeartBeatMonitor

  var waiting_on: String = "client request"
  var current_command: Object = _


  override def create_connection_status = {
    var rc = new OpenwireConnectionStatusDTO
    rc.protocol_version = if (wire_format == null) 0 else wire_format.getVersion
    rc.user = login.map(_.toString).getOrElse(null)
    //    rc.subscription_count = consumers.size
    rc.waiting_on = waiting_on
    rc
  }

  def suspendRead(reason: String) = {
    waiting_on = reason
    connection.transport.suspendRead
  }

  def resumeRead() = {
    waiting_on = "client request"
    connection.transport.resumeRead
  }

  def ack(command: Command):Unit = {
    if (command.isResponseRequired()) {
      val rc = new Response();
      rc.setCorrelationId(command.getCommandId());
      connection_session.offer(rc);
    }
  }


  override def on_transport_failure(error: IOException) = {
    if (!connection.stopped) {
      error.printStackTrace
      suspendRead("shutdown")
      debug(error, "Shutting connection down due to: %s", error)
      connection.stop
    }
  }

  override def on_transport_connected():Unit = {
    security_context.local_address = connection.transport.getLocalAddress
    security_context.remote_address = connection.transport.getRemoteAddress
    outbound_sessions = new SinkMux[Command](connection.transport_sink.map {
      x:Command =>
        x.setCommandId(next_command_id)
        debug("sending frame: %s", x.toString)
        x
    }, dispatchQueue, OpenwireCodec)
    connection_session = new OverflowSink(outbound_sessions.open(dispatchQueue));
    connection_session.refiller = NOOP

    // Send our preferred wire format settings..
    connection.transport.offer(preferred_wireformat_settings)

    resumeRead
    reset {
      suspendRead("virtual host lookup")
      this.host = connection.connector.broker.get_default_virtual_host
      resumeRead
      if(host==null) {
        async_die("Could not find default virtual host")
      }
    }
  }

  override def on_transport_disconnected():Unit = {
    if (!closed) {
      closed = true;
      dead = true;

      heart_beat_monitor.stop<commit_msg>Fix build after recent changes.<commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.activemq.apollo.openwire

import OpenwireConstants._

import org.fusesource.hawtdispatch._
import org.fusesource.hawtbuf._
import collection.mutable.{ListBuffer, HashMap}

import org.apache.activemq.apollo.broker._
import BufferConversions._
import java.io.IOException
import org.apache.activemq.apollo.selector.SelectorParser
import org.apache.activemq.apollo.filter.{BooleanExpression, FilterException}
import org.apache.activemq.apollo.transport._
import org.apache.activemq.apollo.broker.store._
import org.apache.activemq.apollo.util._
import java.util.concurrent.TimeUnit
import java.util.Map.Entry
import protocol._
import scala.util.continuations._
import security.SecurityContext
import tcp.TcpTransport
import codec.OpenWireFormat
import command._
import org.apache.activemq.apollo.openwire.dto.OpenwireConnectionStatusDTO
import org.apache.activemq.apollo.dto.{TopicDestinationDTO, DurableSubscriptionDestinationDTO, DestinationDTO}

object OpenwireProtocolHandler extends Log {

  val DEFAULT_DIE_DELAY = 5 * 1000L
  var die_delay = DEFAULT_DIE_DELAY

  val preferred_wireformat_settings = new WireFormatInfo();
  preferred_wireformat_settings.setVersion(OpenWireFormat.DEFAULT_VERSION);
  preferred_wireformat_settings.setStackTraceEnabled(true);
  preferred_wireformat_settings.setCacheEnabled(true);
  preferred_wireformat_settings.setTcpNoDelayEnabled(true);
  preferred_wireformat_settings.setTightEncodingEnabled(true);
  preferred_wireformat_settings.setSizePrefixDisabled(false);
  preferred_wireformat_settings.setMaxInactivityDuration(30 * 1000 * 1000);
  preferred_wireformat_settings.setMaxInactivityDurationInitalDelay(10 * 1000 * 1000);
  preferred_wireformat_settings.setCacheSize(1024);

}


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

class OpenwireProtocolHandler extends ProtocolHandler {

  var minimum_protocol_version = 1

  import OpenwireProtocolHandler._

  def dispatchQueue: DispatchQueue = connection.dispatch_queue

  def protocol = PROTOCOL

  var outbound_sessions: SinkMux[Command] = null
  var connection_session: Sink[Command] = null
  var closed = false

  var last_command_id=0

  def next_command_id = {
    last_command_id += 1
    last_command_id
  }

  var producerRoutes = new LRUCache[List[DestinationDTO], DeliveryProducerRoute](10) {
    override def onCacheEviction(eldest: Entry[List[DestinationDTO], DeliveryProducerRoute]) = {
      host.router.disconnect(eldest.getKey.toArray, eldest.getValue)
    }
  }

  var host: VirtualHost = null

  private def queue = connection.dispatch_queue

  var session_id: AsciiBuffer = _
  var wire_format: OpenWireFormat = _
  var login: Option[AsciiBuffer] = None
  var passcode: Option[AsciiBuffer] = None
  var dead = false
  val security_context = new SecurityContext

  var heart_beat_monitor: HeartBeatMonitor = new HeartBeatMonitor

  var waiting_on: String = "client request"
  var current_command: Object = _


  override def create_connection_status = {
    var rc = new OpenwireConnectionStatusDTO
    rc.protocol_version = if (wire_format == null) null else wire_format.getVersion.toString
    rc.user = login.map(_.toString).getOrElse(null)
    //    rc.subscription_count = consumers.size
    rc.waiting_on = waiting_on
    rc
  }

  def suspendRead(reason: String) = {
    waiting_on = reason
    connection.transport.suspendRead
  }

  def resumeRead() = {
    waiting_on = "client request"
    connection.transport.resumeRead
  }

  def ack(command: Command):Unit = {
    if (command.isResponseRequired()) {
      val rc = new Response();
      rc.setCorrelationId(command.getCommandId());
      connection_session.offer(rc);
    }
  }


  override def on_transport_failure(error: IOException) = {
    if (!connection.stopped) {
      error.printStackTrace
      suspendRead("shutdown")
      debug(error, "Shutting connection down due to: %s", error)
      connection.stop
    }
  }

  override def on_transport_connected():Unit = {
    security_context.local_address = connection.transport.getLocalAddress
    security_context.remote_address = connection.transport.getRemoteAddress
    outbound_sessions = new SinkMux[Command](connection.transport_sink.map {
      x:Command =>
        x.setCommandId(next_command_id)
        debug("sending openwire command: %s", x.toString())
        x
    }, dispatchQueue, OpenwireCodec)
    connection_session = new OverflowSink(outbound_sessions.open(dispatchQueue));
    connection_session.refiller = NOOP

    // Send our preferred wire format settings..
    connection.transport.offer(preferred_wireformat_settings)

    resumeRead
    reset {
      suspendRead("virtual host lookup")
      this.host = connection.connector.broker.get_default_virtual_host
      resumeRead
      if(host==null) {
        async_die("Could not find default virtual host")
      }
    }
  }

  override def on_transport_disconnected():Unit = {
    if (!closed) {
      closed = true;
      dead = true;

      heart_beat_monitor.stop
{'fraction_non_alphanumeric': 0.08231577149688327, 'fraction_numerical': 0.00657501494321578, 'mean_word_length': 4.367552703941338, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4337794', 'n_tokens_mistral': 3436, 'n_tokens_neox': 3358, 'n_words': 806}