/**
*
*/
+@Deprecated
public final class NeverReconnectStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractNeverReconnectStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class NeverReconnectStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractNeverReconnectStrategyFactoryModuleFactory
{
/**
*
*/
+@Deprecated
public final class ReconnectImmediatelyStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractReconnectImmediatelyStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class ReconnectImmediatelyStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractReconnectImmediatelyStrategyFactoryModuleFactory
{
/**
*
*/
+@Deprecated
public final class TimedReconnectStrategyFactoryModule extends org.opendaylight.controller.config.yang.protocol.framework.AbstractTimedReconnectStrategyFactoryModule
{
/**
*
*/
+@Deprecated
public class TimedReconnectStrategyFactoryModuleFactory extends org.opendaylight.controller.config.yang.protocol.framework.AbstractTimedReconnectStrategyFactoryModuleFactory
{
* Dispatcher class for creating servers and clients. The idea is to first create servers and clients and the run the
* start method that will handle sockets in different thread.
*/
+@Deprecated
public abstract class AbstractDispatcher<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> implements Closeable {
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
public abstract class AbstractProtocolSession<M> extends SimpleChannelInboundHandler<Object> implements ProtocolSession<M> {
private static final Logger LOG = LoggerFactory.getLogger(AbstractProtocolSession.class);
* @param <M> Protocol message type
* @param <S> Protocol session type, has to extend ProtocolSession<M>
*/
+@Deprecated
public abstract class AbstractSessionNegotiator<M, S extends AbstractProtocolSession<?>> extends ChannelInboundHandlerAdapter implements SessionNegotiator<S> {
private final Logger LOG = LoggerFactory.getLogger(AbstractSessionNegotiator.class);
private final Promise<S> promise;
* Utility ReconnectStrategy singleton, which will cause the reconnect process
* to always fail.
*/
+@Deprecated
@ThreadSafe
public final class NeverReconnectStrategy implements ReconnectStrategy {
private final EventExecutor executor;
*
* This interface should be implemented by a final class representing a protocol specific session.
*/
+@Deprecated
public interface ProtocolSession<T> extends Closeable {
@Override
void close();
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
@ThreadSafe
final class ProtocolSessionPromise<S extends ProtocolSession<?>> extends DefaultPromise<S> {
private static final Logger LOG = LoggerFactory.getLogger(ProtocolSessionPromise.class);
* Utility ReconnectStrategy singleton, which will cause the reconnect process
* to immediately schedule a reconnection attempt.
*/
+@Deprecated
@ThreadSafe
public final class ReconnectImmediatelyStrategy implements ReconnectStrategy {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectImmediatelyStrategy.class);
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
final class ReconnectPromise<S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> extends DefaultPromise<Void> {
private static final Logger LOG = LoggerFactory.getLogger(ReconnectPromise.class);
* not attempt any more connection attempts and should abort the reconnection
* process.
*/
+@Deprecated
public interface ReconnectStrategy {
/**
* Query the strategy for the connect timeout.
* primarily useful for allowing injection of a specific type of strategy for
* on-demand use, pretty much like you would use a ThreadFactory.
*/
+@Deprecated
public interface ReconnectStrategyFactory {
/**
* Create a new ReconnectStrategy.
* implemented by a protocol specific abstract class, that is extended by
* a final class that implements the methods.
*/
+@Deprecated
public interface SessionListener<M, S extends ProtocolSession<?>, T extends TerminationReason> extends EventListener {
/**
* Fired when the session was established successfully.
* implemented by a protocol specific abstract class, that is extended by
* a final class that implements the methods.
*/
+@Deprecated
public interface SessionListenerFactory<T extends SessionListener<?, ?, ?>> {
/**
* Returns one session listener
*
* @param <T> Protocol session type.
*/
+@Deprecated
public interface SessionNegotiator<T extends ProtocolSession<?>> extends ChannelInboundHandler {
}
*
* @param <S> session type
*/
+@Deprecated
public interface SessionNegotiatorFactory<M, S extends ProtocolSession<?>, L extends SessionListener<?, ?, ?>> {
/**
* Create a new negotiator attached to a channel, which will notify
/**
* Marker interface for grouping session termination cause.
*/
+@Deprecated
public interface TerminationReason {
/**
*
* Both these caps can be combined, with the strategy giving up as soon as the first one is reached.
*/
+@Deprecated
@ThreadSafe
public final class TimedReconnectStrategy implements ReconnectStrategy {
private static final Logger LOG = LoggerFactory.getLogger(TimedReconnectStrategy.class);
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+
+This program and the accompanying materials are made available under the
+terms of the Eclipse Public License v1.0 which accompanies this distribution,
+and is available at http://www.eclipse.org/legal/epl-v10.html
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>sal-parent</artifactId>
+ <version>1.2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>message-bus-api</artifactId>
+ <name>${project.artifactId}</name>
+
+ <packaging>bundle</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller.model</groupId>
+ <artifactId>model-inventory</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.yangtools.model</groupId>
+ <artifactId>yang-ext</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.yangtools.model</groupId>
+ <artifactId>ietf-topology</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${project.build.directory}/generated-sources/sal</outputBaseDir>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${project.build.directory}/generated-sources/config</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.yang.unified.doc.generator.maven.DocumentationGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>target/site/models</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.8</version>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${project.build.directory}/generated-sources/config</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ <configuration>
+ <instructions>
+ <Export-Package>org.opendaylight.yang.gen.v1.urn.cisco.params.xml.ns.yang.messagebus.*</Export-Package>
+ </instructions>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
--- /dev/null
+module event-aggregator {
+ // FIXME: this module needs to be split up to concepts and API
+ // as the concepts are shared with the other model in this
+ // package.
+ yang-version 1;
+ namespace "urn:cisco:params:xml:ns:yang:messagebus:eventaggregator";
+ prefix "eventaggregator";
+
+ organization "Cisco Systems, Inc.";
+ contact "Robert Gallas";
+
+ description
+ "Module implementing message but RPC.
+
+ Copyright (c)2014 Cisco Systems, Inc. All rights reserved.
+
+ This program and the accompanying materials are made available
+ under the terms of the Eclipse Public License v1.0 which
+ accompanies this distribution, and is available at
+ http://www.eclipse.org/legal/epl-v10.html";
+
+ revision "2014-12-02" {
+ description "Initial revision";
+ }
+
+ typedef pattern {
+ type string {
+ length 1..max;
+ }
+
+ // FIXME: make this a regular expression
+ description "A match pattern. Specifically this is a wildcard pattern.";
+ }
+
+ typedef notification-pattern {
+ type pattern;
+ description
+ "Pattern for matching candidate notification types. This pattern is to be
+ applied against the concatenation of the namespace of the module which
+ defines that particular notification, followed by a single colon, and
+ then followed by notification identifier, as supplied in the argument to
+ the notification statement.";
+ }
+
+ typedef topic-id {
+ type string {
+ length 1..max;
+ }
+ description
+ "A topic identifier. It uniquely defines a topic as seen by the the user
+ of this model's RPCs";
+ }
+
+ // FIXME: we would really like to share instances here, but that requires some sort
+ // of sane reference counting. The reason for sharing is the data path part
+ // of notification delivery -- multiple creators of topics can still share
+ // a single data path.
+ rpc create-topic {
+ description
+ "Create a new topic. A topic is an aggregation of several notification
+ types from a set of nodes. Each successful invocation results in a unique
+ topic being created. The caller is responsible for removing the topic
+ once it is no longer needed.";
+
+ input {
+ leaf notification-pattern {
+ type notification-pattern;
+ mandatory true;
+ description
+ "Pattern matching notification which should be forwarded into this
+ topic.";
+ }
+
+ leaf node-id-pattern {
+ type pattern;
+ mandatory true;
+ description
+ "Pattern for matching candidate event source nodes when looking
+ for contributors to the topic. The pattern will be applied against
+ /network-topology/topology/node/node-id";
+ }
+ }
+
+ output {
+ leaf topic-id {
+ type topic-id;
+ mandatory true;
+ }
+ }
+ }
+
+ rpc destroy-topic {
+ description
+ "Destroy a topic. No further messages will be delivered to it.";
+
+ input {
+ leaf topic-id {
+ type topic-id;
+ mandatory true;
+ }
+ }
+ }
+
+ notification topic-notification {
+ description
+ "Notification of an event occuring on a particular node. This notification
+ acts as an encapsulation for the event being delivered.";
+
+ leaf topic-id {
+ type topic-id;
+ mandatory true;
+ description
+ "Topic to which this event is being delivered.";
+ }
+
+ leaf node-id {
+ // FIXME: should be topology node ID
+ type string;
+ mandatory true;
+ description
+ "Node ID of the node which generated the event.";
+ }
+
+ anyxml payload {
+ mandatory true;
+ description
+ "Encapsulated notification. The format is the XML representation of
+ a notification according to RFC6020 section 7.14.2.";
+ }
+ }
+}
--- /dev/null
+module event-source {
+ yang-version 1;
+ namespace "urn:cisco:params:xml:ns:yang:messagebus:eventsource";
+ prefix "eventsource";
+
+ import event-aggregator { prefix aggr; }
+ import network-topology { prefix nt; revision-date "2013-10-21"; }
+ import opendaylight-inventory {prefix inv; revision-date "2013-08-19"; }
+ import yang-ext {prefix ext; revision-date "2013-07-09"; }
+
+ organization "Cisco Systems, Inc.";
+ contact "Robert Gallas";
+
+ description
+ "Base model for a topology where individual nodes can produce events.
+
+ Module implementing event source topology and encapped notification.
+
+ Copyright (c)2014 Cisco Systems, Inc. All rights reserved.
+
+ This program and the accompanying materials are made available
+ under the terms of the Eclipse Public License v1.0 which
+ accompanies this distribution, and is available at
+ http://www.eclipse.org/legal/epl-v10.html";
+
+ revision "2014-12-02" {
+ description "first revision";
+ }
+
+ // FIXME: expand this
+ typedef join-topic-status {
+ type enumeration {
+ enum up;
+ enum down;
+ }
+ description "Object status";
+ }
+
+ // FIXME: migrate to topology
+ typedef node-ref {
+ type leafref {
+ path "/inv:nodes/inv:node/inv:id";
+ }
+ }
+
+ grouping topology-event-source-type {
+ container topology-event-source {
+ presence "indicates an event source-aware topology";
+ }
+ }
+
+ rpc join-topic {
+ input {
+ leaf node {
+ ext:context-reference "inv:node-context";
+ type "instance-identifier";
+ }
+ leaf topic-id {
+ type aggr:topic-id;
+ description "in current implementation notification-pattern is defined by topic-id.
+ By persisting topic definition we could omit notification-pattern";
+ }
+ leaf notification-pattern {
+ type aggr:notification-pattern;
+ }
+ }
+
+ output {
+ leaf status {
+ type join-topic-status;
+ }
+ }
+ }
+
+ augment "/nt:network-topology/nt:topology/nt:topology-types" {
+ uses topology-event-source-type;
+ }
+
+ augment "/nt:network-topology/nt:topology/nt:node" {
+ when "../../nt:topology-types/topology-event-source";
+ leaf event-source-node {
+ type node-ref;
+ }
+ }
+}
<!-- Clustering -->
<module>sal-remoterpc-connector</module>
+
+ <!-- Message Bus -->
+ <module>messagebus-api</module>
</modules>
<build>
+++ /dev/null
-/*
- * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-
-package org.opendaylight.controller.cluster.raft.base.messages;
-
-/**
- * Internal message by Leader to initiate an install snapshot
- */
-public class InitiateInstallSnapshot {
-}
-
import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
-import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
sendHeartBeat();
return this;
- } else if(message instanceof InitiateInstallSnapshot) {
- installSnapshotIfNeeded();
-
} else if(message instanceof SendInstallSnapshot) {
// received from RaftActor
setSnapshot(Optional.of(((SendInstallSnapshot) message).getSnapshot()));
followerNextIndex, leaderSnapShotIndex, leaderLastIndex
);
}
- actor().tell(new InitiateInstallSnapshot(), actor());
// Send heartbeat to follower whenever install snapshot is initiated.
sendAppendEntriesToFollower(followerActor, followerLogInformation.getNextIndex(),
Collections.<ReplicatedLogEntry>emptyList(), followerId);
+ initiateCaptureSnapshot(followerId, followerNextIndex);
+
} else if(sendHeartbeat) {
//we send an AppendEntries, even if the follower is inactive
// in-order to update the followers timestamp, in case it becomes active again
}
/**
- * An installSnapshot is scheduled at a interval that is a multiple of
- * a HEARTBEAT_INTERVAL. This is to avoid the need to check for installing
- * snapshots at every heartbeat.
- *
+ * /**
* Install Snapshot works as follows
- * 1. Leader sends a InitiateInstallSnapshot message to self
- * 2. Leader then initiates the capture snapshot by sending a CaptureSnapshot message to actor
- * 3. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log
+ * 1. Leader initiates the capture snapshot by sending a CaptureSnapshot message to actor
+ * 2. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log
* and makes a call to Leader's handleMessage , with SendInstallSnapshot message.
- * 4. Leader , picks the snapshot from im-mem ReplicatedLog and sends it in chunks to the Follower
- * 5. On complete, Follower sends back a InstallSnapshotReply.
- * 6. On receipt of the InstallSnapshotReply for the last chunk, Leader marks the install complete for that follower
+ * 3. Leader , picks the snapshot from im-mem ReplicatedLog and sends it in chunks to the Follower
+ * 4. On complete, Follower sends back a InstallSnapshotReply.
+ * 5. On receipt of the InstallSnapshotReply for the last chunk, Leader marks the install complete for that follower
* and replenishes the memory by deleting the snapshot in Replicated log.
- *
+ * 6. If another follower requires a snapshot and a snapshot has been collected (via CaptureSnapshotReply)
+ * then send the existing snapshot in chunks to the follower.
+ * @param followerId
+ * @param followerNextIndex
*/
- private void installSnapshotIfNeeded() {
+ private void initiateCaptureSnapshot(String followerId, long followerNextIndex) {
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: installSnapshotIfNeeded, followers {}", context.getId(), followerToLog.keySet());
+ LOG.debug("{}: initiateCaptureSnapshot, followers {}", context.getId(), followerToLog.keySet());
}
- for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
- final ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
-
- if (followerActor != null) {
- long nextIndex = e.getValue().getNextIndex();
-
- if (!context.getReplicatedLog().isPresent(nextIndex) &&
- context.getReplicatedLog().isInSnapshot(nextIndex)) {
- LOG.info("{}: {} follower needs a snapshot install", context.getId(), e.getKey());
- if (snapshot.isPresent()) {
- // if a snapshot is present in the memory, most likely another install is in progress
- // no need to capture snapshot
- sendSnapshotChunk(followerActor, e.getKey());
-
- } else if (!context.isSnapshotCaptureInitiated()) {
- initiateCaptureSnapshot();
- //we just need 1 follower who would need snapshot to be installed.
- // when we have the snapshot captured, we would again check (in SendInstallSnapshot)
- // who needs an install and send to all who need
- break;
- }
+ if (!context.getReplicatedLog().isPresent(followerNextIndex) &&
+ context.getReplicatedLog().isInSnapshot(followerNextIndex)) {
+ if (snapshot.isPresent()) {
+ // if a snapshot is present in the memory, most likely another install is in progress
+ // no need to capture snapshot.
+ // This could happen if another follower needs an install when one is going on.
+ final ActorSelection followerActor = context.getPeerActorSelection(followerId);
+ sendSnapshotChunk(followerActor, followerId);
+
+ } else if (!context.isSnapshotCaptureInitiated()) {
+
+ LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", context.getId(), getLeaderId());
+ ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied());
+ long lastAppliedIndex = -1;
+ long lastAppliedTerm = -1;
+
+ if (lastAppliedEntry != null) {
+ lastAppliedIndex = lastAppliedEntry.getIndex();
+ lastAppliedTerm = lastAppliedEntry.getTerm();
+ } else if (context.getReplicatedLog().getSnapshotIndex() > -1) {
+ lastAppliedIndex = context.getReplicatedLog().getSnapshotIndex();
+ lastAppliedTerm = context.getReplicatedLog().getSnapshotTerm();
}
- }
- }
- }
-
- // on every install snapshot, we try to capture the snapshot.
- // Once a capture is going on, another one issued will get ignored by RaftActor.
- private void initiateCaptureSnapshot() {
- LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", context.getId(), getLeaderId());
- ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied());
- long lastAppliedIndex = -1;
- long lastAppliedTerm = -1;
- if (lastAppliedEntry != null) {
- lastAppliedIndex = lastAppliedEntry.getIndex();
- lastAppliedTerm = lastAppliedEntry.getTerm();
- } else if (context.getReplicatedLog().getSnapshotIndex() > -1) {
- lastAppliedIndex = context.getReplicatedLog().getSnapshotIndex();
- lastAppliedTerm = context.getReplicatedLog().getSnapshotTerm();
+ boolean isInstallSnapshotInitiated = true;
+ actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(),
+ lastAppliedIndex, lastAppliedTerm, isInstallSnapshotInitiated),
+ actor());
+ context.setSnapshotCaptureInitiated(true);
+ }
}
-
- boolean isInstallSnapshotInitiated = true;
- actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(),
- lastAppliedIndex, lastAppliedTerm, isInstallSnapshotInitiated),
- actor());
- context.setSnapshotCaptureInitiated(true);
}
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
-import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot;
+import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.Leader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
assertEquals(5, leaderActor.getReplicatedLog().size());
// simulate a real snapshot
- leaderActor.onReceiveCommand(new InitiateInstallSnapshot());
+ leaderActor.onReceiveCommand(new SendHeartBeat());
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(String.format("expected to be Leader but was %s. Current Leader = %s ",
leaderActor.getCurrentBehavior().state(), leaderActor.getLeaderId())
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
-import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.IsolatedLeaderCheck;
import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
assertTrue(raftBehavior instanceof Leader);
- // we might receive some heartbeat messages, so wait till we InitiateInstallSnapshot
+ // we might receive some heartbeat messages, so wait till we get CaptureSnapshot
Boolean[] matches = new ReceiveWhile<Boolean>(Boolean.class, duration("2 seconds")) {
@Override
protected Boolean match(Object o) throws Exception {
- if (o instanceof InitiateInstallSnapshot) {
+ if (o instanceof CaptureSnapshot) {
return true;
}
return false;
}
}.get();
- boolean initiateInitiateInstallSnapshot = false;
+ boolean captureSnapshot = false;
for (Boolean b: matches) {
- initiateInitiateInstallSnapshot = b | initiateInitiateInstallSnapshot;
+ captureSnapshot = b | captureSnapshot;
}
- assertTrue(initiateInitiateInstallSnapshot);
+ assertTrue(captureSnapshot);
}};
}
ActorRef followerActor = getTestActor();
Map<String, String> peerAddresses = new HashMap<>();
- peerAddresses.put(followerActor.path().toString(),
- followerActor.path().toString());
-
+ peerAddresses.put(followerActor.path().toString(), followerActor.path().toString());
- MockRaftActorContext actorContext =
- (MockRaftActorContext) createActorContext(leaderActor);
+ MockRaftActorContext actorContext = (MockRaftActorContext) createActorContext(leaderActor);
actorContext.setPeerAddresses(peerAddresses);
Map<String, String> leadersSnapshot = new HashMap<>();
leader.setSnapshot(Optional.<ByteString>absent());
// new entry
- ReplicatedLogImplEntry entry =
- new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
+ ReplicatedLogImplEntry entry = new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
new MockRaftActorContext.MockPayload("D"));
actorContext.getReplicatedLog().append(entry);
- // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
+ //update follower timestamp
+ leader.markFollowerActive(followerActor.path().toString());
+
RaftActorBehavior raftBehavior = leader.handleMessage(
- leaderActor, new InitiateInstallSnapshot());
+ senderActor, new Replicate(null, "state-id", entry));
CaptureSnapshot cs = MessageCollectorActor.
getFirstMatching(leaderActor, CaptureSnapshot.class);
assertEquals(2, cs.getLastTerm());
// if an initiate is started again when first is in progress, it shouldnt initiate Capture
- raftBehavior = leader.handleMessage(leaderActor, new InitiateInstallSnapshot());
+ leader.handleMessage(senderActor, new Replicate(null, "state-id", entry));
List<Object> captureSnapshots = MessageCollectorActor.getAllMatching(leaderActor, CaptureSnapshot.class);
assertEquals("CaptureSnapshot should not get invoked when initiate is in progress", 1, captureSnapshots.size());
*/
package org.opendaylight.controller.md.sal.dom.api;
+import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import java.util.Objects;
import javax.annotation.Nonnull;
@Override
public final String toString() {
- return com.google.common.base.Objects.toStringHelper(this).omitNullValues().add("type", type).add("contextReference", getContextReference()).toString();
+ return MoreObjects.toStringHelper(this).omitNullValues().add("type", type).add("contextReference", getContextReference()).toString();
}
}
package org.opendaylight.controller.sal.core.api;
import java.util.concurrent.Future;
-
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcService} instead.
+ */
+@Deprecated
public interface RpcConsumptionRegistry {
/**
* Sends an RPC to other components registered to the broker.
*/
package org.opendaylight.controller.sal.core.api;
+import com.google.common.util.concurrent.ListenableFuture;
import java.util.Set;
-
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
-import com.google.common.util.concurrent.ListenableFuture;
-
/**
* {@link Provider}'s implementation of an RPC.
*
* {@link RpcResult}
* <li> {@link Broker} returns the {@link RpcResult} to {@link Consumer}
* </ol>
+ *
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementation} instead.
*/
+@Deprecated
public interface RpcImplementation extends Provider.ProviderFunctionality {
/**
/**
* Exception reported when no RPC implementation is found in the system.
+ *
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException} instead.
*/
+@Deprecated
public class RpcImplementationUnavailableException extends RuntimeException {
private static final long serialVersionUID = 1L;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcProviderService} and {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcService} instead.
+ */
+@Deprecated
public interface RpcProvisionRegistry extends RpcImplementation, BrokerService, RouteChangePublisher<RpcRoutingContext, YangInstanceIdentifier>, DOMService {
/**
package org.opendaylight.controller.sal.core.api;
import java.util.EventListener;
-
import org.opendaylight.yangtools.yang.common.QName;
+/**
+ * @deprecated Use {@link org.opendaylight.controller.md.sal.dom.api.DOMRpcAvailabilityListener} instead.
+ */
+@Deprecated
public interface RpcRegistrationListener extends EventListener {
public void onRpcImplementationAdded(QName name);
*/
package org.opendaylight.controller.md.sal.dom.broker.impl;
-import com.google.common.base.Objects;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FutureCallback;
@Override
public String toString() {
- return addToStringAttributes(Objects.toStringHelper(this)).toString();
+ return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
}
protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.japi.Creator;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
private final Configuration configuration;
private final String followerId;
private final Logger LOG = LoggerFactory.getLogger(DummyShard.class);
+ private long lastMessageIndex = -1;
+ private long lastMessageSize = 0;
public DummyShard(Configuration configuration, String followerId) {
this.configuration = configuration;
}
protected void handleAppendEntries(AppendEntries req) throws InterruptedException {
- LOG.info("{} - Received AppendEntries message : leader term, index, size = {}, {}, {}", followerId, req.getTerm(),req.getLeaderCommit(), req.getEntries().size());
+
+ LOG.info("{} - Received AppendEntries message : leader term = {}, index = {}, prevLogIndex = {}, size = {}",
+ followerId, req.getTerm(),req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
+
+ if(lastMessageIndex == req.getLeaderCommit() && req.getEntries().size() > 0 && lastMessageSize > 0){
+ LOG.error("{} - Duplicate message with leaderCommit = {} prevLogIndex = {} received", followerId, req.getLeaderCommit(), req.getPrevLogIndex());
+ }
+
+ lastMessageIndex = req.getLeaderCommit();
+ lastMessageSize = req.getEntries().size();
+
long lastIndex = req.getLeaderCommit();
- if (req.getEntries().size() > 0)
- lastIndex = req.getEntries().get(0).getIndex();
+ if (req.getEntries().size() > 0) {
+ for(ReplicatedLogEntry entry : req.getEntries()) {
+ lastIndex = entry.getIndex();
+ }
+ }
- if (configuration.shouldCauseTrouble()) {
+ if (configuration.shouldCauseTrouble() && req.getEntries().size() > 0) {
boolean ignore = false;
if (configuration.shouldDropReplies()) {
--- /dev/null
+org.slf4j.simpleLogger.showDateTime=true
+org.slf4j.simpleLogger.dateTimeFormat=hh:mm:ss,S a
+org.slf4j.simpleLogger.logFile=System.out
+org.slf4j.simpleLogger.showShortLogName=true
+org.slf4j.simpleLogger.levelInBrackets=true
+org.slf4j.simpleLogger.defaultLogLevel=info
\ No newline at end of file
*/
package org.opendaylight.controller.md.sal.dom.store.impl;
-import com.google.common.base.Objects;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Preconditions;
-
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
import org.slf4j.Logger;
@Override
public final String toString() {
- return addToStringAttributes(Objects.toStringHelper(this)).toString();
+ return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
}
/**
package org.opendaylight.controller.md.sal.dom.store.impl;
import static com.google.common.base.Preconditions.checkState;
-import com.google.common.base.Objects.ToStringHelper;
+import com.google.common.base.MoreObjects.ToStringHelper;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;