GitHub PullRequestReviewCommentEvent at 2014-01-07 10:53:09
Repository: akka
Actor: patriknw
Payload: {"comment":{"url":"https://api.github.com/repos/akka/akka/pulls/comments/8690258","id":8690258,"diff_hunk":"@@ -0,0 +1,131 @@\n+/**\n+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>\n+ */\n+\n+package akka.persistence\n+\n+import scala.concurrent.duration._\n+\n+import akka.actor._\n+import akka.persistence.JournalProtocol._\n+\n+/**\n+ * Instructs a replicator to run a replication (i.e. an incremental message replay).\n+ *\n+ * @param awaitReplication if `true` processing of further messages will be delayed until\n+ *                         replication completes. If `false`, any following message may\n+ *                         interleave with the replayed [[Persistent]] message stream.\n+ */\n+case class Replicate(awaitReplication: Boolean)\n+\n+/**\n+ * A replicator replicates the persistent message stream of a corresponding processor. Messages written\n+ * by a processor are automatically received by all replicators with a matching `processorId`. There is\n+ * no direct communication between a processor its replicators. Instead, replicators execute incremental\n+ * [[Persistent]] message replays by interacting with the journal. Replicators update themselves using a\n+ * pull-based approach.\n+ *\n+ * Replication runs periodically with an interval that can be configured via the\n+ * `akka.persistence.replicator.default-replication-interval` key or by overriding `replicationInterval`.\n+ * Overriding the `replicationInterval` method takes precedence. An incremental replication can also be\n+ * triggered by applications by sending a replicator a [[Replicate]] message.\n+ *\n+ * Replicators can be used to generate (eventual consistent) views of their corresponding processor's\n+ * state (the R in CQRS). These views can also run on different nodes, provided that a replicated journal\n+ * is used.\n+ *\n+ * Replicators can also store snapshots of internal state by calling [[saveSnapshot]]. During recovery,\n+ * a saved snapshot is offered to the replicator with a [[SnapshotOffer]] message, followed by replayed\n+ * messages, if any, that are younger than the snapshot. Default is to offer the latest saved snapshot.\n+ */\n+trait Replicator extends Actor with Recovery {\n+  import context.dispatcher\n+\n+  /**\n+   * INTERNAL API.\n+   *\n+   * Extends the `recoveryStarted` state of [[Recovery]] with logic to handle [[Replicate]] requests sent\n+   * by users.\n+   */\n+  private[persistence] override def recoveryStarted(awaitReplay: Boolean) = new State {\n+    private var fallback = Replicator.super.recoveryStarted(awaitReplay)\n+\n+    override def aroundReceive(receive: Receive, message: Any) = message match {\n+      case Replicate(awaitReplication) ⇒ if (!awaitReplay && awaitReplication)\n+        fallback = Replicator.super.recoveryStarted(awaitReplay = true)\n+      case other ⇒\n+        fallback.aroundReceive(receive, other)\n+    }\n+  }\n+\n+  /**\n+   * Schedules a [[Replicate]] message when this state is entered. Runs an incremental replay if [[Replicate]]\n+   * is received, otherwise invokes the actor's behavior with any other received message.\n+   */\n+  private def idle: State = new State {\n+    override def toString: String = \"idle\"\n+\n+    schedule = Some(context.system.scheduler.scheduleOnce(replicationInterval, self, Replicate(awaitReplication = false)))\n+\n+    def aroundReceive(receive: Receive, message: Any): Unit = message match {\n+      case Replicate(awaitReplication) ⇒\n+        _currentState = recoveryStarted(awaitReplay = awaitReplication)\n+        journal ! Replay(lastSequenceNr + 1L, Long.MaxValue, processorId, self)\n+      case r: Recover ⇒ // ignore\n+      case other      ⇒ process(receive, other)\n+    }\n+  }\n+\n+  /**\n+   * INTERNAL API.\n+   */\n+  private[persistence] def onReplaySuccess(receive: Receive, await: Boolean, maxSnr: Long): Unit =\n+    onReplayComplete(await)\n+\n+  /**\n+   * INTERNAL API.\n+   */\n+  private[persistence] def onReplayFailure(receive: Receive, await: Boolean, cause: Throwable): Unit =\n+    onReplayComplete(await)\n+\n+  private val defaultReplicationInterval = FiniteDuration(configuredReplicationInterval, MILLISECONDS)\n+  private val _replicatorId = extension.processorId(self)\n+\n+  private var schedule: Option[Cancellable] = None\n+\n+  /**\n+   * Replicator id. Defaults to this replicator's path and can be overridden.\n+   */\n+  def replicatorId: String = _replicatorId\n+\n+  /**\n+   * Returns `replicatorId`.\n+   */\n+  def snapshotterId: String = replicatorId\n+\n+  /**\n+   * This replicator's replication interval. It defaults to 5 seconds and can be changed by\n+   * configuration via the `akka.persistence.replicator.default-replication-interval` key or\n+   * overriding this method. Overriding takes precedence over configuration.\n+   */\n+  def replicationInterval: FiniteDuration =","path":"akka-persistence/src/main/scala/akka/persistence/Replicator.scala","position":111,"original_position":111,"commit_id":"b2185c50e79dac538926f879e88688b08e1296c9","original_commit_id":"b2185c50e79dac538926f879e88688b08e1296c9","user":{"login":"patriknw","id":336161,"avatar_url":"https://gravatar.com/avatar/a036dbdf494ae4c9d46e1d24b53fd5da?d=https%3A%2F%2Fidenticons.github.com%2F21e8602cb97e90820d8888ed8fe77c3c.png&r=x","gravatar_id":"a036dbdf494ae4c9d46e1d24b53fd5da","url":"https://api.github.com/users/patriknw","html_url":"https://github.com/patriknw","followers_url":"https://api.github.com/users/patriknw/followers","following_url":"https://api.github.com/users/patriknw/following{/other_user}","gists_url":"https://api.github.com/users/patriknw/gists{/gist_id}","starred_url":"https://api.github.com/users/patriknw/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/patriknw/subscriptions","organizations_url":"https://api.github.com/users/patriknw/orgs","repos_url":"https://api.github.com/users/patriknw/repos","events_url":"https://api.github.com/users/patriknw/events{/privacy}","received_events_url":"https://api.github.com/users/patriknw/received_events","type":"User","site_admin":false},"body":"this should use the setting in the Persistence extension","created_at":"2014-01-07T10:53:09Z","updated_at":"2014-01-07T10:53:09Z","html_url":"https://github.com/akka/akka/pull/1905#discussion_r8690258","pull_request_url":"https://api.github.com/repos/akka/akka/pulls/1905","_links":{"self":{"href":"https://api.github.com/repos/akka/akka/pulls/comments/8690258"},"html":{"href":"https://github.com/akka/akka/pull/1905#discussion_r8690258"},"pull_request":{"href":"https://api.github.com/repos/akka/akka/pulls/1905"}}}}