<commit_before>/**
 * Copyright (C) 2010-2011 LShift Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.lshift.diffa.kernel.actors

import java.util.concurrent.TimeUnit.MILLISECONDS
import org.slf4j.{Logger, LoggerFactory}
import net.jcip.annotations.ThreadSafe
import java.util.concurrent.ScheduledFuture
import net.lshift.diffa.kernel.differencing._
import net.lshift.diffa.kernel.events.{VersionID, PairChangeEvent}
import net.lshift.diffa.kernel.participants.{DownstreamParticipant, UpstreamParticipant}
import org.joda.time.DateTime
import net.lshift.diffa.kernel.util.AlertCodes
import com.eaio.uuid.UUID
import akka.actor._
import collection.mutable.{SynchronizedQueue, Queue}
import concurrent.SyncVar
import net.lshift.diffa.kernel.diag.{DiagnosticLevel, DiagnosticsManager}
import net.lshift.diffa.kernel.util.StoreSynchronizationUtils._
import net.lshift.diffa.kernel.config.{Endpoint, DiffaPair}
import net.lshift.diffa.participant.scanning.{ScanResultEntry, ScanConstraint}

/**
 * This actor serializes access to the underlying version policy from concurrent processes.
 */
case class PairActor(pair:DiffaPair,
                     us:Endpoint,
                     ds:Endpoint,
                     usp:UpstreamParticipant,
                     dsp:DownstreamParticipant,
                     policy:VersionPolicy,
                     store:VersionCorrelationStore,
                     differencesManager:DifferencesManager,
                     pairScanListener:PairScanListener,
                     diagnostics:DiagnosticsManager,
                     changeEventBusyTimeoutMillis: Long,
                     changeEventQuietTimeoutMillis: Long) extends Actor {

  val logger:Logger = LoggerFactory.getLogger(getClass)

  self.id_=(pair.identifier)

  private val pairRef = pair.asRef

  private var lastEventTime: Long = 0
  private var scheduledFlushes: ScheduledFuture[_] = _

  /**
   * Flag that can be used to signal that scanning should be cancelled.
   */
  @ThreadSafe
  private var feedbackHandle:FeedbackHandle = null

  lazy val writer = store.openWriter()

  /**
   * A queue of deferred messages that arrived during a scanning state
   */
  private val deferred = new Queue[Deferrable]

  /**
   * Detail of the current scan-in-progress. Contains a UUID that lets us identify operations that belong to it.
   */
  private var activeScan:OutstandingScan = null

  /**
   * Keep track of scans that are still outstanding. This lets us know whether messages that arrive
   * are entirely spurious, or just jobs cleaning themselves up.
   */
  private val outstandingScans = collection.mutable.Map[UUID, OutstandingScan]()

  /**
   * This is the address of the client that requested the last cancellation
   */
  private var cancellationRequester:Channel[Any] = null

  /**
   * This allows tracing of spurious messages, but is only enabled in when the log level is set to TRACE
   */
  abstract class TraceableCommand(uuid:UUID) {
    var exception:Throwable = null
    if (logger.isTraceEnabled) {
      exception = new Exception().fillInStackTrace()
    }
  }

  /**
   * Describes a message coming from a child actor running as part of a scan
   */
  trait ChildActorScanMessage {
    def scanUuid:UUID   // The uuid of the scan that this message is coming from
  }

  private case class OutstandingScan(uuid:UUID) {
    var upstreamCompleted = false
    var downstreamCompleted = false

    def isCompleted = upstreamCompleted && downstreamCompleted
  }

  /**
   * This is the set of commands that the writer proxy understands
   */
  case class VersionCorrelationWriterCommand(scanUuid:UUID, invokeWriter:(LimitedVersionCorrelationWriter => Correlation))
      extends TraceableCommand(scanUuid)
      with ChildActorScanMessage {
    override def toString = scanUuid.toString
  }

  /**
   * Marker messages to let the actor know that a portion of the scan has successfully completed.
   */
  case class ChildActorCompletionMessage(scanUuid:UUID, upOrDown:UpOrDown, result:Result)
      extends ChildActorScanMessage {
    def logMessage(l:Logger, s:ActorState, code:String)
      = l.debug("%s: UUID[%s] -> Received %sstream %s in %s state".format(code, scanUuid, upOrDown, result, s))
  }

  /**
   * This proxy is presented to clients that need access to a LimitedVersionCorrelationWriter.
   * It wraps the underlying writer instance and forwards all commands via asynchronous messages,
   * thus allowing parallel access to the writer.
   */
  private def createWriterProxy(scanUuid:UUID) = new LimitedVersionCorrelationWriter() {
    // The receive timeout in seconds
    val timeout = 60

    def clearUpstreamVersion(id: VersionID) = call( _.clearUpstreamVersion(id) )
    def clearDownstreamVersion(id: VersionID) = call( _.clearUpstreamVersion(id) )
    def storeDownstreamVersion(id: VersionID, attributes: Map[String, TypedAttribute], lastUpdated: DateTime, uvsn: String, dvsn: String)
      = call( _.storeDownstreamVersion(id, attributes, lastUpdated, uvsn, dvsn) )
    def storeUpstreamVersion(id: VersionID, attributes: Map[String, TypedAttribute], lastUpdated: DateTime, vsn: String)
      = call( _.storeUpstreamVersion(id, attributes, lastUpdated, vsn) )
    def call(command:(LimitedVersionCorrelationWriter => Correlation)) = {
      val message = VersionCorrelationWriterCommand(scanUuid, command)<commit_msg>Fixed writer proxy to call clearDownstreamVersion on the underlying writer<commit_after>/**
 * Copyright (C) 2010-2011 LShift Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.lshift.diffa.kernel.actors

import java.util.concurrent.TimeUnit.MILLISECONDS
import org.slf4j.{Logger, LoggerFactory}
import net.jcip.annotations.ThreadSafe
import java.util.concurrent.ScheduledFuture
import net.lshift.diffa.kernel.differencing._
import net.lshift.diffa.kernel.events.{VersionID, PairChangeEvent}
import net.lshift.diffa.kernel.participants.{DownstreamParticipant, UpstreamParticipant}
import org.joda.time.DateTime
import net.lshift.diffa.kernel.util.AlertCodes
import com.eaio.uuid.UUID
import akka.actor._
import collection.mutable.{SynchronizedQueue, Queue}
import concurrent.SyncVar
import net.lshift.diffa.kernel.diag.{DiagnosticLevel, DiagnosticsManager}
import net.lshift.diffa.kernel.util.StoreSynchronizationUtils._
import net.lshift.diffa.kernel.config.{Endpoint, DiffaPair}
import net.lshift.diffa.participant.scanning.{ScanResultEntry, ScanConstraint}

/**
 * This actor serializes access to the underlying version policy from concurrent processes.
 */
case class PairActor(pair:DiffaPair,
                     us:Endpoint,
                     ds:Endpoint,
                     usp:UpstreamParticipant,
                     dsp:DownstreamParticipant,
                     policy:VersionPolicy,
                     store:VersionCorrelationStore,
                     differencesManager:DifferencesManager,
                     pairScanListener:PairScanListener,
                     diagnostics:DiagnosticsManager,
                     changeEventBusyTimeoutMillis: Long,
                     changeEventQuietTimeoutMillis: Long) extends Actor {

  val logger:Logger = LoggerFactory.getLogger(getClass)

  self.id_=(pair.identifier)

  private val pairRef = pair.asRef

  private var lastEventTime: Long = 0
  private var scheduledFlushes: ScheduledFuture[_] = _

  /**
   * Flag that can be used to signal that scanning should be cancelled.
   */
  @ThreadSafe
  private var feedbackHandle:FeedbackHandle = null

  lazy val writer = store.openWriter()

  /**
   * A queue of deferred messages that arrived during a scanning state
   */
  private val deferred = new Queue[Deferrable]

  /**
   * Detail of the current scan-in-progress. Contains a UUID that lets us identify operations that belong to it.
   */
  private var activeScan:OutstandingScan = null

  /**
   * Keep track of scans that are still outstanding. This lets us know whether messages that arrive
   * are entirely spurious, or just jobs cleaning themselves up.
   */
  private val outstandingScans = collection.mutable.Map[UUID, OutstandingScan]()

  /**
   * This is the address of the client that requested the last cancellation
   */
  private var cancellationRequester:Channel[Any] = null

  /**
   * This allows tracing of spurious messages, but is only enabled in when the log level is set to TRACE
   */
  abstract class TraceableCommand(uuid:UUID) {
    var exception:Throwable = null
    if (logger.isTraceEnabled) {
      exception = new Exception().fillInStackTrace()
    }
  }

  /**
   * Describes a message coming from a child actor running as part of a scan
   */
  trait ChildActorScanMessage {
    def scanUuid:UUID   // The uuid of the scan that this message is coming from
  }

  private case class OutstandingScan(uuid:UUID) {
    var upstreamCompleted = false
    var downstreamCompleted = false

    def isCompleted = upstreamCompleted && downstreamCompleted
  }

  /**
   * This is the set of commands that the writer proxy understands
   */
  case class VersionCorrelationWriterCommand(scanUuid:UUID, invokeWriter:(LimitedVersionCorrelationWriter => Correlation))
      extends TraceableCommand(scanUuid)
      with ChildActorScanMessage {
    override def toString = scanUuid.toString
  }

  /**
   * Marker messages to let the actor know that a portion of the scan has successfully completed.
   */
  case class ChildActorCompletionMessage(scanUuid:UUID, upOrDown:UpOrDown, result:Result)
      extends ChildActorScanMessage {
    def logMessage(l:Logger, s:ActorState, code:String)
      = l.debug("%s: UUID[%s] -> Received %sstream %s in %s state".format(code, scanUuid, upOrDown, result, s))
  }

  /**
   * This proxy is presented to clients that need access to a LimitedVersionCorrelationWriter.
   * It wraps the underlying writer instance and forwards all commands via asynchronous messages,
   * thus allowing parallel access to the writer.
   */
  private def createWriterProxy(scanUuid:UUID) = new LimitedVersionCorrelationWriter() {
    // The receive timeout in seconds
    val timeout = 60

    def clearUpstreamVersion(id: VersionID) = call( _.clearUpstreamVersion(id) )
    def clearDownstreamVersion(id: VersionID) = call( _.clearDownstreamVersion(id) )
    def storeDownstreamVersion(id: VersionID, attributes: Map[String, TypedAttribute], lastUpdated: DateTime, uvsn: String, dvsn: String)
      = call( _.storeDownstreamVersion(id, attributes, lastUpdated, uvsn, dvsn) )
    def storeUpstreamVersion(id: VersionID, attributes: Map[String, TypedAttribute], lastUpdated: DateTime, vsn: String)
      = call( _.storeUpstreamVersion(id, attributes, lastUpdated, vsn) )
    def call(command:(LimitedVersionCorrelationWriter => Correlation)) = {
      val message = VersionCorrelationWriterCommand(scanUuid, command)
{'fraction_non_alphanumeric': 0.0778498609823911, 'fraction_numerical': 0.00269609908164125, 'mean_word_length': 3.6476115896632733, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '1464095', 'n_tokens_mistral': 3255, 'n_tokens_neox': 3153, 'n_words': 1026}