*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public abstract class Action implements Serializable {
private static final long serialVersionUID = 1L;
private static final Logger logger = LoggerFactory.getLogger(Action.class);
* The enumeration of actions supported by the controller
* Each entry has a unique id and the values range for the action element where applicable
*/
+@Deprecated
public enum ActionType {
DROP("drop", 0, 0),
LOOPBACK("loopback", 0, 0),
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Controller extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Drop extends Action {
private static final long serialVersionUID = 1L;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Enqueue extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Flood extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class FloodAll extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class HwPath extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Loopback extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Output extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class PopVlan extends Action {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class PushVlan extends Action {
private static final long serialVersionUID = 1L;
private int tag; // TPID - 16 bits
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetDlDst extends Action {
private static final long serialVersionUID = 1L;
private byte[] address;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetDlSrc extends Action {
private static final long serialVersionUID = 1L;
private byte[] address;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetDlType extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetNextHop extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetNwDst extends Action {
private static final long serialVersionUID = 1L;
InetAddress address;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetNwSrc extends Action {
private static final long serialVersionUID = 1L;
InetAddress address;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetNwTos extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetTpDst extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetTpSrc extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetVlanCfi extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetVlanId extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SetVlanPcp extends Action {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SupportedFlowActions extends Property {
private static final long serialVersionUID = 1L;
public static final String SupportedFlowActionsPropName = "supportedFlowActions";
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SwPath extends Action {
private static final long serialVersionUID = 1L;
* It contains the role name and the role level in the
* application context as specified by {@link AppRoleLevel}
*/
+@Deprecated
public class AppRole implements Serializable {
private static final long serialVersionUID = 1L;
String name;
* In the controller space such a role will be seen as <code>APPUSER<code>
* as specified in {@link UserLevel}
*/
+@Deprecated
public enum AppRoleLevel implements Serializable {
APPADMIN(0, "App-Admin", "Application Administrator"), APPUSER(1,
"App-User", "Application User"), APPOPERATOR(2, "App-Operator",
import java.io.Serializable;
+@Deprecated
public enum AuthResultEnum implements Serializable {
AUTH_NONE("AUTH_NOT_ATTEMPTED"), AUTH_ACCEPT("AUTHENTICATION_ACCEPTED"), // request accepted
AUTH_REJECT("AUTHENTICATION_REJECTED"), // request rejected
* and User Manager make use of this interface to retrieve
* authorization information at user or and role level.
*/
+@Deprecated
public interface IResourceAuthorization {
/**
/**
* It represents the group/resource access privilege
*/
+@Deprecated
public enum Privilege implements Serializable {
NONE(""), // no privilege
READ("r"), // read only
* It represents the elementary resource along with
* the access privilege associated to it
*/
+@Deprecated
public class Resource implements Serializable {
private static final long serialVersionUID = 1L;
Object resource; // the generic resource
*
*
*/
+@Deprecated
public class ResourceGroup implements Serializable {
private static final long serialVersionUID = 1L;
private String groupName; // the resource group name
/**
* Describes the user role level in the controller space
*/
+@Deprecated
public enum UserLevel implements Serializable {
SYSTEMADMIN(0, "System-Admin", "System Administrator"), // can do everything
NETWORKADMIN(1, "Network-Admin", "Network Administrator"), // can do everything but setting a system admin user profile
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
@SuppressWarnings("serial")
+@Deprecated
public class AdvertisedBandwidth extends Bandwidth {
public static final String AdvertisedBandwidthPropName = "advertisedBandwidth";
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Bandwidth extends Property {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Buffers extends Property {
private static final long serialVersionUID = 1L;
@XmlElement(name="value")
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Capabilities extends Property {
private static final long serialVersionUID = 1L;
@XmlElement(name="value")
* register dependencies per-container
*
*/
+@Deprecated
abstract public class ComponentActivatorAbstractBase implements
BundleActivator, IContainerAware {
Logger logger = LoggerFactory
@XmlRootElement
@SuppressWarnings("serial")
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Config extends Property {
@XmlElement(name="value")
private short configValue;
*/
package org.opendaylight.controller.sal.core;
+@Deprecated
public class ConstructionException extends Exception {
private static final long serialVersionUID = 1L;
*
*
*/
+@Deprecated
public class ContainerFlow implements Serializable {
private static final long serialVersionUID = 1L;
private Match match;
* Class representing a ServiceDependency on a container
*
*/
+@Deprecated
public class ContainerServiceDependency implements ServiceDependency,
DependencyActivation {
private ServiceDependency m_dep;
@XmlRootElement
@SuppressWarnings("serial")
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Description extends Property {
@XmlElement(name="value")
private String descriptionValue;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Edge implements Serializable {
private static final long serialVersionUID = 1L;
@XmlElement
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
@SuppressWarnings("serial")
+@Deprecated
public class ForwardingMode extends Property {
@XmlElement(name="value")
private final int modeValue;
@XmlRootElement(name="host")
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Host implements Serializable {
private static final long serialVersionUID = 1L;
@XmlElement
*
* Interface used to retrieve the status of a given Container
*/
+@Deprecated
public interface IContainer {
/**
* Returns the Name of the container described
* created/destroyed
*/
+@Deprecated
public interface IContainerAware {
/**
* Method invoked to signal that a container is being created
* The interface describes methods used to retrieve the status of a given
* Container
*/
+@Deprecated
public interface IContainerListener {
/**
* Called to notify a change in the tag assigned to a switch
* The interface describes methods used to publish the changes to a given
* Container configuration to listeners on the local cluster node only.
*/
+@Deprecated
public interface IContainerLocalListener extends IContainerListener {
}
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Latency extends Property {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class MacAddress extends Property implements Cloneable {
private static final long serialVersionUID = 1L;
@XmlElement(name="value")
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
@SuppressWarnings("serial")
+@Deprecated
public class Name extends Property {
@XmlElement(name="value")
private String nameValue;
*/
@XmlAccessorType(XmlAccessType.NONE)
@XmlRootElement
+@Deprecated
public class Node implements Serializable {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NodeConnector implements Serializable {
private static final long serialVersionUID = 1L;
public static final Short SPECIALNODECONNECTORID = (short) 0;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NodeTable implements Serializable {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Path implements Serializable {
private static final long serialVersionUID = 1L;
@XmlElement
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class PeerBandwidth extends Bandwidth {
private static final long serialVersionUID = 1L;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
abstract public class Property implements Serializable, Cloneable {
private static final long serialVersionUID = 1L;
private final String name;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
@SuppressWarnings("serial")
+@Deprecated
public class State extends Property {
@XmlElement(name="value")
private short stateValue;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class SupportedBandwidth extends Bandwidth {
private static final long serialVersionUID = 1L;
public static final String SupportedBandwidthPropName = "supportedBandwidth";
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Tables extends Property {
private static final long serialVersionUID = 1L;
@XmlElement(name="value")
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
@SuppressWarnings("serial")
+@Deprecated
public class Tier extends Property {
@XmlElement(name="value")
private int tierValue;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class TimeStamp extends Property {
private static final long serialVersionUID = 1L;
@XmlElement(name = "value")
* @brief Describes update types
*
*/
+@Deprecated
public enum UpdateType {
ADDED("added"), REMOVED("removed"), CHANGED("changed");
* The interface provides the methods to notify the listener when an edge is
* added/deleted/changed.
*/
+@Deprecated
public interface IDiscoveryService {
/**
* The methods is called when an edge is added/deleted/changed
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Flow implements Cloneable, Serializable {
protected static final Logger logger = LoggerFactory.getLogger(Flow.class);
private static final long serialVersionUID = 1L;
* functional modules the asynchronous messages related to flow programming
* coming from the network nodes.
*/
+@Deprecated
public interface IFlowProgrammerListener extends IPluginOutFlowProgrammerService {
}
* Interface that defines the methods available to the functional modules above
* SAL for installing/modifying/removing flows on a network node
*/
+@Deprecated
public interface IFlowProgrammerService {
/**
* Synchronously add a flow to the network node
* This interface defines the flow programmer methods to be implemented by
* protocol plugins
*/
+@Deprecated
public interface IPluginInFlowProgrammerService {
/**
* Synchronously add a flow to the network node
* inform the SAL layer about the asynchronous messages related to flow
* programming coming from the network nodes.
*/
+@Deprecated
public interface IPluginOutFlowProgrammerService {
/**
* Inform SAL that the flow on the specified node has been removed Consumer
* The Interface describes methods invoked from application toward SAL to
* solicit existing inventory data.
*/
+@Deprecated
public interface IInventoryService {
/**
* The method retrieves all the existing nodes and properties attached
* The interface provides the methods to notify the upper applications in
* regards to any inventory changes.
*/
+@Deprecated
public interface IListenInventoryUpdates {
/**
* This method is called when some properties of a node are added/deleted/changed.
* The Interface describes methods invoked from SAL toward the protocol plugin
* to solicit existing inventory data.
*/
+@Deprecated
public interface IPluginInInventoryService {
/**
* The method retrieves all the existing nodes and properties attached
* The Interface describes Inventory update methods to be implemented by
* protocol plugin.
*/
+@Deprecated
public interface IPluginOutInventoryService {
/**
* This method is called when some properties of a node are added/deleted/changed.
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Match implements Cloneable, Serializable {
private static final long serialVersionUID = 1L;
private static final Map<MatchType, MatchType> reversableMatches;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class MatchField implements Cloneable, Serializable {
private static final long serialVersionUID = 1L;
private static final Logger logger = LoggerFactory.getLogger(MatchField.class);
* values of the elements type that can be matched on the network
* frame/packet/message
*/
+@Deprecated
public enum MatchType {
IN_PORT("inPort", 1 << 0, NodeConnector.class, 1, 0),
DL_SRC("dlSrc", 1 << 1, Byte[].class, 0, 0xffffffffffffL),
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class DlDst extends MatchField<byte[]> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "DL_DST";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class DlSrc extends MatchField<byte[]> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "DL_SRC";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class DlType extends MatchField<Short> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "DL_TYPE";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class DlVlan extends MatchField<Short> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "DL_VLAN";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class DlVlanPriority extends MatchField<Byte> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "DL_VLAN_PR";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class InPort extends MatchField<NodeConnector> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "IN_PORT";
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class Match implements Cloneable, Serializable {
private static final long serialVersionUID = 1L;
private Map<String, MatchField<?>> fields;
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public abstract class MatchField<T> implements Cloneable, Serializable {
private static final long serialVersionUID = 1L;
private String type;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NwDst extends MatchField<InetAddress> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "NW_DST";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NwProtocol extends MatchField<Byte> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "NW_PROTO";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NwSrc extends MatchField<InetAddress> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "NW_SRC";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NwTos extends MatchField<Byte> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "NW_TOS";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class TpDst extends MatchField<Short> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "TP_DST";
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class TpSrc extends MatchField<Short> {
private static final long serialVersionUID = 1L;
public static final String TYPE = "TP_SRC";
*
*
*/
-
+@Deprecated
public class ARP extends Packet {
private static final String HWTYPE = "HardwareType";
private static final String PTYPE = "ProtocolType";
* - store bits in specified location in stream of bits
* - convert primitive data types to stream of bits
*/
+@Deprecated
public abstract class BitBufferHelper {
protected static final Logger logger = LoggerFactory
.getLogger(BitBufferHelper.class);
/**
* Describes an exception that is raised during BitBufferHelper operations.
*/
+@Deprecated
public class BufferException extends Exception {
private static final long serialVersionUID = 1L;
/**
* Class that represents the Ethernet frame objects
*/
+@Deprecated
public class Ethernet extends Packet {
private static final String DMAC = "DestinationMACAddress";
private static final String SMAC = "SourceMACAddress";
/**
* Class that represents the ICMP packet objects
*/
-
+@Deprecated
public class ICMP extends Packet {
private static final String TYPE = "Type";
private static final String CODE = "Code";
/**
* Data Packet Services SAL provides to the components
*/
+@Deprecated
public interface IDataPacketService {
/**
* Transmit a data Packet. Transmission will ONLY happen if the
/**
* Class that represents the IEEE 802.1Q objects
*/
+@Deprecated
public class IEEE8021Q extends Packet {
private static final String PCP = "PriorityCodePoint";
private static final String CFI = "CanonicalFormatIndicator";
* incoming DataPacket match the filter. If no filter is provided, the
* handler is called for EVERY packet i.e. match All is implied!
*/
+@Deprecated
public interface IListenDataPacket {
/**
* Handler for receiving the packet. The application can signal
* - "protocoloPluginType"
* the value of the property will org.opendaylight.controller.sal.core.Node.NodeIDType
*/
+@Deprecated
public interface IPluginInDataPacketService {
/**
* Transmit a data Packet. Packet will go out ONLY if the packet
* Interface used by SAL to intercept any Data Packet coming from the
* southbound protocol plugins
*/
+@Deprecated
public interface IPluginOutDataPacketService {
/**
* Handler for receiving the packet. The SAL layer can signal back
/**
* Class that represents the IPv4 packet objects
*/
-
+@Deprecated
public class IPv4 extends Packet {
protected static final Logger logger = LoggerFactory
.getLogger(IPv4.class);
/**
* Class that represents the LLDP frame objects
*/
-
+@Deprecated
public class LLDP extends Packet {
private static final String CHASSISID = "ChassisId";
private static final String SYSTEMNAMEID = "SystemNameID";
/**
* Class that represents the LLDPTLV objects
*/
-
+@Deprecated
public class LLDPTLV extends Packet {
private static final String TYPE = "Type";
private static final String LENGTH = "Length";
/**
* Describe a packet data link format
*/
+@Deprecated
public enum LinkEncap {
ETHERNET
}
* the basic methods which are common for all the packets, like serialize and
* deserialize
*/
-
+@Deprecated
public abstract class Packet {
protected static final Logger logger = LoggerFactory
.getLogger(Packet.class);
* packet/stream is malformed.
*
*/
+@Deprecated
public class PacketException extends Exception {
private static final long serialVersionUID = 1L;
* Possible results for Data packet processing handler
*
*/
+@Deprecated
public enum PacketResult {
/**
* Packet has been processed and noone in the chain after us is
* and how it will be transmitted. It essentially wraps the raw bytestream
*
*/
+@Deprecated
public class RawPacket {
private byte[] packetData;
private final LinkEncap encap;
/**
* Class that represents the TCP segment objects
*/
+@Deprecated
public class TCP extends Packet {
public static final String SRCPORT = "SourcePort";
/**
* Class that represents the UDP datagram objects
*/
-
+@Deprecated
public class UDP extends Packet {
private static final String SRCPORT = "SourcePort";
*
*/
@XmlRootElement
+@Deprecated
abstract public class DataLinkAddress implements Serializable {
private static final long serialVersionUID = 1L;
private String name;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class EthernetAddress extends DataLinkAddress {
private static final long serialVersionUID = 1L;
@XmlTransient
@XmlRootElement (name="FlowStat")
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class FlowOnNode implements Serializable{
private static final long serialVersionUID = 1L;
/**
* The interface defines hardware view read methods to be implemented by protocol plugins
*/
+@Deprecated
public interface IPluginInReadService {
/**
* The interface defines hardware statistics updates service to be offered by
* protocol plugins
*/
+@Deprecated
public interface IPluginOutReadService {
/**
* This interface defines methods for retrieving the network node's
* flow/port/queue hardware view
*/
+@Deprecated
public interface IReadService {
/**
* Get the hardware view for the specified flow on the specified network
* The interface describes SAL service to be consumed by functional modules that
* are interested in reader updates
*/
+@Deprecated
public interface IReadServiceListener extends IPluginOutReadService {
}
*/
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NodeConnectorStatistics implements Serializable {
private static final long serialVersionUID = 1L;
/**
* Represents the network node description information
*/
+@Deprecated
public class NodeDescription implements Serializable, Cloneable{
private static final long serialVersionUID = 1L;
@XmlRootElement
@XmlAccessorType(XmlAccessType.NONE)
+@Deprecated
public class NodeTableStatistics implements Serializable {
private static final long serialVersionUID = 1L;
* know events published by the routing engine
*
*/
-
+@Deprecated
public interface IListenRoutingUpdates {
/**
* Method invoked when the recalculation of the all shortest path
* This interface provides APIs to manage and query the routing information
*
*/
+@Deprecated
public interface IRouting {
/**
* registry. This interface (on a per-container base) and SAL will call it
* providing the update.
*/
+@Deprecated
public interface IListenTopoUpdates {
/**
* Called to update on Edge in the topology graph
* plugin to sollicit Topoloy updates
*
*/
+@Deprecated
public interface IPluginInTopologyService {
/**
* Tell to protocol plugin that is time to send the updates of the
* this service provided by SAL so the update can migrate upward toward the
* applications.
*/
+@Deprecated
public interface IPluginOutTopologyService {
/**
/**
* Topology methods provided by SAL toward the applications
*/
+@Deprecated
public interface ITopologyService {
/**
* Tell to SAL that is time to send the updates of the
* The class represents an Edge, the Edge's Property Set and its UpdateType.
* If update is on new properties added to an existing edge, appropriate type is CHANGED.
*/
+@Deprecated
public class TopoEdgeUpdate {
private Edge edge;
private Set<Property> props;
* This interface defines the methods for configuration object
*
*/
+@Deprecated
public interface ConfigurationObject {
}
*
*
*/
+@Deprecated
public enum Direction implements Serializable {
FORWARD("forward"), REVERSE("reverse");
private Direction(String name) {
*
*
*/
+@Deprecated
public enum EtherTypes {
PVSTP("PVSTP", 0x010B), // 802.2 + SNAP (Spanning Tree)
CDP("CDP", 0x2000), // 802.2 + SNAP
* An iterator that will filter values from an iterator and return only those
* values that match the predicate.
*/
+@Deprecated
public abstract class FilterIterator<T> implements Iterator<T> {
protected Iterator<T> subIterator;
protected T next;
*
*
*/
+@Deprecated
public enum GUIField {
USER("User"), PASSWORD("Password"), ROLE("Role"), SERVERIP("Server address"), SERVERSECRET(
"Server secret"), SERVERPROTOCOL("Server protocol"), NAME("Name"), CONTAINER(
* Global Constants
*
*/
+@Deprecated
public enum GlobalConstants {
DEFAULT("default"),
CONTAINERMANAGER("containermanager"),
*
*
*/
+@Deprecated
public class HexEncode {
/**
* This method converts byte array into String format without ":" inserted.
* This interface defines the methods for callback ordering
*
*/
-
+@Deprecated
public interface IListener<T> {
public enum Command {
CONTINUE, STOP
* This interface defines the methods to be called when looking up custom node types
*
*/
-
+@Deprecated
public interface INodeConnectorFactory {
/**
* Method to get custom NodeConnector types from protocol plugins
* This interface defines the methods to be called when looking up custom node types
*
*/
-
+@Deprecated
public interface INodeFactory {
/**
* Method to get custom node types from protocol plugins
*
*
*/
+@Deprecated
public interface IObjectReader {
public Object readObject(ObjectInputStream ois)
throws FileNotFoundException, IOException, ClassNotFoundException;
* references:
* http://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
*/
+@Deprecated
public enum IPProtocols {
ANY("any", -1),
HOPOPT("HOPOPT",0),
* @param <T>
* the type of elements returned by this iterator
*/
+@Deprecated
public class IterableIterator<T> implements Iterator<T> {
Iterator<? extends Iterable<T>> subIterator;
Iterator<T> current = null;
* @author readams
*
*/
+@Deprecated
public class ListenerDispatcher<U, T extends IListener<U>> {
protected static Logger logger = LoggerFactory
.getLogger(ListenerDispatcher.class);
* @param <T>
* the type of elements returned by this iterator
*/
+@Deprecated
public class MultiIterator<T> implements Iterator<T> {
Iterator<Iterator<T>> subIterator;
Iterator<T> current = null;
* Utility class containing the common utility functions needed for operating on
* networking data structures
*/
+@Deprecated
public abstract class NetUtils {
protected static final Logger logger = LoggerFactory.getLogger(NetUtils.class);
/**
*
*
*/
+@Deprecated
public abstract class NodeConnectorCreator {
protected static final Logger logger = LoggerFactory
.getLogger(NodeConnectorCreator.class);
*
*
*/
+@Deprecated
public abstract class NodeCreator {
protected static final Logger logger = LoggerFactory.getLogger(NodeCreator.class);
import org.opendaylight.controller.sal.core.NodeTable.NodeTableIDType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
+@Deprecated
public class NodeTableCreator {
protected static final Logger logger = LoggerFactory
.getLogger(NodeTableCreator.class);
*
*
*/
+@Deprecated
public class ObjectReader {
private static Logger logger = LoggerFactory.getLogger(ObjectReader.class);
private FileInputStream fis;
* Write object to write to file stream
*
*/
+@Deprecated
public class ObjectWriter {
private static Logger logger = LoggerFactory.getLogger(ObjectWriter.class);
private FileOutputStream fos;
*
*
*/
+@Deprecated
public class ReadFromFile {
private FileInputStream fileStream;
private File filePointer;
*
*
*/
+@Deprecated
public class ServiceHelper {
private static final Logger logger = LoggerFactory
.getLogger(ServiceHelper.class);
* new task * If the task has begun, set a bit to restart it after the current
* task finishes
*/
+@Deprecated
public class SingletonTask {
protected static Logger logger = LoggerFactory
.getLogger(SingletonTask.class);
* It contains a code {@code StatusCode} representing the result of the call and
* a string which describes a failure reason (if any) in human readable form.
*/
+@Deprecated
public class Status implements Serializable {
private static final long serialVersionUID = 0L;
private StatusCode code;
* Each enum value is associated with a minimal description string.
*
*/
+@Deprecated
public enum StatusCode {
SUCCESS("Success"),
CREATED("Created"),
import java.util.ArrayList;
import java.util.List;
-
+@Deprecated
public final class TierHelper {
private static TierHelper tierHelper;
public static final int unknownTierNumber = 0;
*
*
*/
+@Deprecated
public class WriteToFile {
protected static final Logger logger = LoggerFactory
.getLogger(WriteToFile.class);
<scope>runtime</scope>
</dependency>
<dependency>
- <groupId>${symbol_dollar}{groupId}</groupId>
+ <groupId>${symbol_dollar}{project.groupId}</groupId>
<artifactId>${artifactId}-impl</artifactId>
<version>${symbol_dollar}{project.version}</version>
</dependency>
<dependency>
- <groupId>${symbol_dollar}{groupId}</groupId>
+ <groupId>${symbol_dollar}{project.groupId}</groupId>
<artifactId>${artifactId}-api</artifactId>
<version>${symbol_dollar}{project.version}</version>
</dependency>
<packaging>bundle</packaging>
<dependencies>
<dependency>
- <groupId>${symbol_dollar}{groupId}</groupId>
+ <groupId>${symbol_dollar}{project.groupId}</groupId>
<artifactId>${artifactId}-api</artifactId>
<version>${symbol_dollar}{project.version}</version>
</dependency>
<type>xml</type>
<scope>runtime</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>features-restconf</artifactId>
+ <version>${project.version}</version>
+ <classifier>features</classifier>
+ <type>xml</type>
+ <scope>runtime</scope>
+ </dependency>
<!-- FIXME: move this into netconf-artifacts -->
<dependency>
</dependencies>
</dependencyManagement>
</project>
-
private final DataPersistenceProvider dataPersistenceProvider;
private long persistIdentifier = 1;
- private Optional<ActorRef> roleChangeNotifier;
+ private
final Optional<ActorRef> roleChangeNotifier;
public ExampleActor(String id, Map<String, String> peerAddresses,
} catch (Exception e) {
LOG.error(e, "Exception in creating snapshot");
}
- getSelf().tell(new CaptureSnapshotReply(bs), null);
+ getSelf().tell(new CaptureSnapshotReply(bs.toByteArray()), null);
}
- @Override protected void applySnapshot(ByteString snapshot) {
+ @Override protected void applySnapshot(byte [] snapshot) {
state.clear();
try {
state.putAll((HashMap) toObject(snapshot));
}
}
- private Object toObject(ByteString bs) throws ClassNotFoundException, IOException {
+ private Object toObject(byte [] bs) throws ClassNotFoundException, IOException {
Object obj = null;
ByteArrayInputStream bis = null;
ObjectInputStream ois = null;
try {
- bis = new ByteArrayInputStream(bs.toByteArray());
+ bis = new ByteArrayInputStream(bs);
ois = new ObjectInputStream(bis);
obj = ois.readObject();
} finally {
}
@Override
- protected void applyRecoverySnapshot(ByteString snapshot) {
+ protected void applyRecoverySnapshot(byte[] snapshot) {
}
}
timer.start();
// Apply the snapshot to the actors state
- applyRecoverySnapshot(ByteString.copyFrom(snapshot.getState()));
+ applyRecoverySnapshot(snapshot.getState());
timer.stop();
LOG.info("Recovery snapshot applied for {} in {}: snapshotIndex={}, snapshotTerm={}, journal-size=" +
snapshot.getLastAppliedTerm()
);
}
- applySnapshot(ByteString.copyFrom(snapshot.getState()));
+
+ applySnapshot(snapshot.getState());
//clears the followers log, sets the snapshot index to ensure adjusted-index works
replicatedLog = new ReplicatedLogImpl(snapshot);
} else if (message instanceof CaptureSnapshot) {
LOG.info("CaptureSnapshot received by actor");
- CaptureSnapshot cs = (CaptureSnapshot)message;
- captureSnapshot = cs;
- createSnapshot();
- } else if (message instanceof CaptureSnapshotReply){
- LOG.info("CaptureSnapshotReply received by actor");
- CaptureSnapshotReply csr = (CaptureSnapshotReply) message;
+ if(captureSnapshot == null) {
+ captureSnapshot = (CaptureSnapshot)message;
+ createSnapshot();
+ }
- ByteString stateInBytes = csr.getSnapshot();
- LOG.info("CaptureSnapshotReply stateInBytes size:{}", stateInBytes.size());
- handleCaptureSnapshotReply(stateInBytes);
+ } else if (message instanceof CaptureSnapshotReply){
+ handleCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot());
} else {
if (!(message instanceof AppendEntriesMessages.AppendEntries)
*
* @param snapshot A snapshot of the state of the actor
*/
- protected abstract void applyRecoverySnapshot(ByteString snapshot);
+ protected abstract void applyRecoverySnapshot(byte[] snapshotBytes);
/**
* This method is called during recovery at the end of a batch to apply the current batched
* operations when the derived actor is out of sync with it's peers
* and the only way to bring it in sync is by applying a snapshot
*
- * @param snapshot A snapshot of the state of the actor
+ * @param snapshotBytes A snapshot of the state of the actor
*/
- protected abstract void applySnapshot(ByteString snapshot);
+ protected abstract void applySnapshot(byte[] snapshotBytes);
/**
* This method will be called by the RaftActor when the state of the
return peerAddress;
}
- private void handleCaptureSnapshotReply(ByteString stateInBytes) {
+ private void handleCaptureSnapshotReply(byte[] snapshotBytes) {
+ LOG.info("CaptureSnapshotReply received by actor: snapshot size {}", snapshotBytes.length);
+
// create a snapshot object from the state provided and save it
// when snapshot is saved async, SaveSnapshotSuccess is raised.
- Snapshot sn = Snapshot.create(stateInBytes.toByteArray(),
+ Snapshot sn = Snapshot.create(snapshotBytes,
context.getReplicatedLog().getFrom(captureSnapshot.getLastAppliedIndex() + 1),
captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
if (isLeader() && captureSnapshot.isInstallSnapshotInitiated()) {
// this would be call straight to the leader and won't initiate in serialization
- currentBehavior.handleMessage(getSelf(), new SendInstallSnapshot(stateInBytes));
+ currentBehavior.handleMessage(getSelf(), new SendInstallSnapshot(
+ ByteString.copyFrom(snapshotBytes)));
}
captureSnapshot = null;
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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;
+
+/**
+ * @author Thomas Pantelis
+ */
+public interface RaftVersions {
+ short HELIUM_VERSION = 0;
+ short LITHIUM_VERSION = 1;
+ short CURRENT_VERSION = LITHIUM_VERSION;
+}
import java.io.Serializable;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-public class ReplicatedLogImplEntry implements ReplicatedLogEntry,
- Serializable {
- private static final long serialVersionUID = 1L;
+public class ReplicatedLogImplEntry implements ReplicatedLogEntry, Serializable {
+ private static final long serialVersionUID = -9085798014576489130L;
private final long index;
private final long term;
this.payload = payload;
}
- @Override public Payload getData() {
+ @Override
+ public Payload getData() {
return payload;
}
- @Override public long getTerm() {
+ @Override
+ public long getTerm() {
return term;
}
- @Override public long getIndex() {
+ @Override
+ public long getIndex() {
return index;
}
return getData().size();
}
- @Override public String toString() {
+ @Override
+ public String toString() {
return "Entry{" +
"index=" + index +
", term=" + term +
public class SerializationUtils {
public static Object fromSerializable(Object serializable){
- if(serializable.getClass().equals(AppendEntries.SERIALIZABLE_CLASS)){
+ if(AppendEntries.isSerializedType(serializable)){
return AppendEntries.fromSerializable(serializable);
} else if (serializable.getClass().equals(InstallSnapshot.SERIALIZABLE_CLASS)) {
*/
package org.opendaylight.controller.cluster.raft.base.messages;
-import com.google.protobuf.ByteString;
public class CaptureSnapshotReply {
- private ByteString snapshot;
+ private final byte [] snapshot;
- public CaptureSnapshotReply(ByteString snapshot) {
+ public CaptureSnapshotReply(byte [] snapshot) {
this.snapshot = snapshot;
}
- public ByteString getSnapshot() {
+ public byte [] getSnapshot() {
return snapshot;
}
-
- public void setSnapshot(ByteString snapshot) {
- this.snapshot = snapshot;
- }
}
package org.opendaylight.controller.cluster.raft.messages;
public class AbstractRaftRPC implements RaftRPC {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = -6061342433962854822L;
+
// term
protected long term;
public AbstractRaftRPC() {
}
+ @Override
public long getTerm() {
return term;
}
package org.opendaylight.controller.cluster.raft.messages;
import com.google.protobuf.GeneratedMessage;
-import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
-import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
-import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
-
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import org.opendaylight.controller.cluster.raft.RaftVersions;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
+import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
+import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
/**
* Invoked by leader to replicate log entries (§5.3); also used as
* heartbeat (§5.2).
*/
public class AppendEntries extends AbstractRaftRPC {
- public static final Class<AppendEntriesMessages.AppendEntries> SERIALIZABLE_CLASS = AppendEntriesMessages.AppendEntries.class;
+ private static final long serialVersionUID = 1L;
+
+ public static final Class<AppendEntriesMessages.AppendEntries> LEGACY_SERIALIZABLE_CLASS =
+ AppendEntriesMessages.AppendEntries.class;
private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(AppendEntries.class);
- private static final long serialVersionUID = 1L;
// So that follower can redirect clients
private final String leaderId;
// log entries to store (empty for heartbeat;
// may send more than one for efficiency)
- private final List<ReplicatedLogEntry> entries;
+ private transient List<ReplicatedLogEntry> entries;
// leader's commitIndex
private final long leaderCommit;
this.leaderCommit = leaderCommit;
}
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.writeShort(RaftVersions.CURRENT_VERSION);
+ out.defaultWriteObject();
+
+ out.writeInt(entries.size());
+ for(ReplicatedLogEntry e: entries) {
+ out.writeObject(e);
+ }
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.readShort(); // version
+
+ in.defaultReadObject();
+
+ int size = in.readInt();
+ entries = new ArrayList<>(size);
+ for(int i = 0; i < size; i++) {
+ entries.add((ReplicatedLogEntry) in.readObject());
+ }
+ }
+
public String getLeaderId() {
return leaderId;
}
return leaderCommit;
}
- @Override public String toString() {
+ @Override
+ public String toString() {
final StringBuilder sb =
new StringBuilder("AppendEntries{");
sb.append("term=").append(getTerm());
return sb.toString();
}
- public <T extends Object> Object toSerializable(){
+ public <T extends Object> Object toSerializable() {
+ return toSerializable(RaftVersions.CURRENT_VERSION);
+ }
+
+ public <T extends Object> Object toSerializable(short version) {
+ if(version < RaftVersions.LITHIUM_VERSION) {
+ return toLegacySerializable();
+ } else {
+ return this;
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private <T> Object toLegacySerializable() {
AppendEntriesMessages.AppendEntries.Builder to = AppendEntriesMessages.AppendEntries.newBuilder();
to.setTerm(this.getTerm())
.setLeaderId(this.getLeaderId())
return to.build();
}
- public static AppendEntries fromSerializable(Object o){
- AppendEntriesMessages.AppendEntries from = (AppendEntriesMessages.AppendEntries) o;
+ public static AppendEntries fromSerializable(Object serialized) {
+ if(serialized instanceof AppendEntries) {
+ return (AppendEntries)serialized;
+ }
+ else {
+ return fromLegacySerializable((AppendEntriesMessages.AppendEntries) serialized);
+ }
+ }
+ private static AppendEntries fromLegacySerializable(AppendEntriesMessages.AppendEntries from) {
List<ReplicatedLogEntry> logEntryList = new ArrayList<>();
for (AppendEntriesMessages.AppendEntries.ReplicatedLogEntry leProtoBuff : from.getLogEntriesList()) {
String clientPayloadClassName = leProtoBuff.getData().getClientPayloadClassName();
payload = (Payload) Class.forName(clientPayloadClassName).newInstance();
payload = payload.decode(leProtoBuff.getData());
- payload.setClientPayloadClassName(clientPayloadClassName);
} else {
LOG.error("Payload is null or payload does not have client payload class name");
}
return to;
}
+
+ public static boolean isSerializedType(Object message) {
+ return message instanceof AppendEntries || LEGACY_SERIALIZABLE_CLASS.isInstance(message);
+ }
}
private long currentTerm = 0;
private String votedFor = "";
+ @Override
public long getCurrentTerm() {
return currentTerm;
}
+ @Override
public String getVotedFor() {
return votedFor;
}
+ @Override
public void update(long currentTerm, String votedFor){
this.currentTerm = currentTerm;
this.votedFor = votedFor;
return lastApplied;
}
+ @Override
public void setReplicatedLog(ReplicatedLog replicatedLog) {
this.replicatedLog = replicatedLog;
}
}
public static class MockPayload extends Payload implements Serializable {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 3121380393130864247L;
private String value = "";
public MockPayload(){
return MockPayload.class.getName();
}
+ @Override
public String toString() {
return value;
}
}
public static class MockReplicatedLogBuilder {
- private ReplicatedLog mockLog = new SimpleReplicatedLog();
+ private final ReplicatedLog mockLog = new SimpleReplicatedLog();
public MockReplicatedLogBuilder createEntries(int start, int end, int term) {
for (int i=start; i<end; i++) {
package org.opendaylight.controller.cluster.raft;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import scala.concurrent.duration.FiniteDuration;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyObject;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
public class RaftActorTest extends AbstractActorTest {
}
@Override
- protected void applyRecoverySnapshot(ByteString snapshot) {
- delegate.applyRecoverySnapshot(snapshot);
+ protected void applyRecoverySnapshot(byte[] bytes) {
+ delegate.applyRecoverySnapshot(bytes);
try {
- Object data = toObject(snapshot);
- System.out.println("!!!!!applyRecoverySnapshot: "+data);
+ Object data = toObject(bytes);
if (data instanceof List) {
state.addAll((List<?>) data);
}
delegate.createSnapshot();
}
- @Override protected void applySnapshot(ByteString snapshot) {
+ @Override protected void applySnapshot(byte [] snapshot) {
delegate.applySnapshot(snapshot);
}
return this.getId();
}
- private Object toObject(ByteString bs) throws ClassNotFoundException, IOException {
+ private Object toObject(byte[] bs) throws ClassNotFoundException, IOException {
Object obj = null;
ByteArrayInputStream bis = null;
ObjectInputStream ois = null;
try {
- bis = new ByteArrayInputStream(bs.toByteArray());
+ bis = new ByteArrayInputStream(bs);
ois = new ObjectInputStream(bis);
obj = ois.readObject();
} finally {
mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
- verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes));
+ verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes.toByteArray()));
mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
- verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(ByteString.class));
+ verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(byte[].class));
mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
- mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
verify(dataPersistenceProvider).saveSnapshot(anyObject());
verify(mockRaftActor.delegate).createSnapshot();
- mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
mockRaftActor.onReceiveCommand(new SaveSnapshotSuccess(new SnapshotMetadata("foo", 100, 100)));
mockRaftActor.onReceiveCommand(new ApplySnapshot(snapshot));
- verify(mockRaftActor.delegate).applySnapshot(eq(snapshotBytes));
+ verify(mockRaftActor.delegate).applySnapshot(eq(snapshot.getState()));
assertTrue("The replicatedLog should have changed",
oldReplicatedLog != mockRaftActor.getReplicatedLog());
mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1));
- mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
mockRaftActor.onReceiveCommand(new SaveSnapshotFailure(new SnapshotMetadata("foobar", 10L, 1234L),
new Exception()));
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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;
+
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit tests for ReplicatedLogImplEntry.
+ *
+ * @author Thomas Pantelis
+ */
+public class ReplicatedLogImplEntryTest {
+
+ @Test
+ public void testBackwardsCompatibleDeserializationFromHelium() throws Exception {
+ String expPayloadData = "This is a test";
+ int expIndex = 1;
+ int expTerm = 2;
+
+ try(FileInputStream fis = new FileInputStream("src/test/resources/helium-serialized-ReplicatedLogImplEntry")) {
+ ObjectInputStream ois = new ObjectInputStream(fis);
+
+ ReplicatedLogImplEntry entry = (ReplicatedLogImplEntry) ois.readObject();
+ ois.close();
+
+ Assert.assertEquals("getIndex", expIndex, entry.getIndex());
+ Assert.assertEquals("getTerm", expTerm, entry.getTerm());
+
+ MockRaftActorContext.MockPayload payload = (MockRaftActorContext.MockPayload) entry.getData();
+ Assert.assertEquals("data", expPayloadData, payload.toString());
+ }
+ }
+
+ /**
+ * Use this method to generate a file with a serialized ReplicatedLogImplEntry instance to be
+ * used in tests that verify backwards compatible de-serialization.
+ */
+ private void generateSerializedFile() throws IOException {
+ String expPayloadData = "This is a test";
+ int expIndex = 1;
+ int expTerm = 2;
+
+ ReplicatedLogImplEntry entry = new ReplicatedLogImplEntry(expIndex, expTerm,
+ new MockRaftActorContext.MockPayload(expPayloadData));
+ FileOutputStream fos = new FileOutputStream("src/test/resources/serialized-ReplicatedLogImplEntry");
+ ObjectOutputStream oos = new ObjectOutputStream(fos);
+ oos.writeObject(entry);
+ fos.close();
+ }
+}
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages;
import scala.concurrent.duration.FiniteDuration;
public class LeaderTest extends AbstractRaftActorBehaviorTest {
- private ActorRef leaderActor =
+ private final ActorRef leaderActor =
getSystem().actorOf(Props.create(DoNothingActor.class));
- private ActorRef senderActor =
+ private final ActorRef senderActor =
getSystem().actorOf(Props.create(DoNothingActor.class));
@Test
leader.handleMessage(leaderActor, new SendHeartBeat());
- AppendEntriesMessages.AppendEntries aeproto = (AppendEntriesMessages.AppendEntries)MessageCollectorActor.getFirstMatching(
- followerActor, AppendEntries.SERIALIZABLE_CLASS);
+ AppendEntries aeproto = (AppendEntries)MessageCollectorActor.getFirstMatching(
+ followerActor, AppendEntries.class);
assertNotNull("AppendEntries should be sent even if InstallSnapshotReply is not " +
"received", aeproto);
leader.handleMessage(leaderActor, new SendHeartBeat());
- AppendEntriesMessages.AppendEntries appendEntries =
- (AppendEntriesMessages.AppendEntries) MessageCollectorActor
- .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class);
+ AppendEntries appendEntries = (AppendEntries) MessageCollectorActor
+ .getFirstMatching(followerActor, AppendEntries.class);
assertNotNull(appendEntries);
assertEquals(1, appendEntries.getLeaderCommit());
- assertEquals(1, appendEntries.getLogEntries(0).getIndex());
+ assertEquals(1, appendEntries.getEntries().get(0).getIndex());
assertEquals(0, appendEntries.getPrevLogIndex());
AppendEntriesReply appendEntriesReply =
leader.handleMessage(leaderActor, new SendHeartBeat());
- AppendEntriesMessages.AppendEntries appendEntries =
- (AppendEntriesMessages.AppendEntries) MessageCollectorActor
- .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class);
+ AppendEntries appendEntries = (AppendEntries) MessageCollectorActor
+ .getFirstMatching(followerActor, AppendEntries.class);
assertNotNull(appendEntries);
assertEquals(1, appendEntries.getLeaderCommit());
- assertEquals(1, appendEntries.getLogEntries(0).getIndex());
+ assertEquals(1, appendEntries.getEntries().get(0).getIndex());
assertEquals(0, appendEntries.getPrevLogIndex());
AppendEntriesReply appendEntriesReply =
private class MockConfigParamsImpl extends DefaultConfigParamsImpl {
- private long electionTimeOutIntervalMillis;
- private int snapshotChunkSize;
+ private final long electionTimeOutIntervalMillis;
+ private final int snapshotChunkSize;
public MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) {
super();
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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.messages;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import org.apache.commons.lang.SerializationUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
+import org.opendaylight.controller.cluster.raft.RaftVersions;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
+import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
+
+/**
+ * Unit tests for AppendEntries.
+ *
+ * @author Thomas Pantelis
+ */
+public class AppendEntriesTest {
+
+ @Test
+ public void testSerialization() {
+ ReplicatedLogEntry entry1 = new ReplicatedLogImplEntry(1, 2, new MockPayload("payload1"));
+
+ ReplicatedLogEntry entry2 = new ReplicatedLogImplEntry(3, 4, new MockPayload("payload2"));
+
+ AppendEntries expected = new AppendEntries(5L, "node1", 7L, 8L, Arrays.asList(entry1, entry2), 10L);
+
+ AppendEntries cloned = (AppendEntries) SerializationUtils.clone(expected);
+
+ verifyAppendEntries(expected, cloned);
+ }
+
+ @Test
+ public void testToAndFromSerializable() {
+ AppendEntries entries = new AppendEntries(5L, "node1", 7L, 8L,
+ Collections.<ReplicatedLogEntry>emptyList(), 10L);
+
+ assertSame("toSerializable", entries, entries.toSerializable());
+ assertSame("fromSerializable", entries,
+ org.opendaylight.controller.cluster.raft.SerializationUtils.fromSerializable(entries));
+ }
+
+ @Test
+ public void testToAndFromLegacySerializable() {
+ ReplicatedLogEntry entry = new ReplicatedLogImplEntry(3, 4, new MockPayload("payload"));
+ AppendEntries entries = new AppendEntries(5L, "node1", 7L, 8L, Arrays.asList(entry), 10L);
+
+ Object serializable = entries.toSerializable(RaftVersions.HELIUM_VERSION);
+ Assert.assertTrue(serializable instanceof AppendEntriesMessages.AppendEntries);
+
+ AppendEntries entries2 = (AppendEntries)
+ org.opendaylight.controller.cluster.raft.SerializationUtils.fromSerializable(serializable);
+
+ verifyAppendEntries(entries, entries2);
+ }
+
+ private void verifyAppendEntries(AppendEntries expected, AppendEntries actual) {
+ assertEquals("getLeaderId", expected.getLeaderId(), actual.getLeaderId());
+ assertEquals("getTerm", expected.getTerm(), actual.getTerm());
+ assertEquals("getLeaderCommit", expected.getLeaderCommit(), actual.getLeaderCommit());
+ assertEquals("getPrevLogIndex", expected.getPrevLogIndex(), actual.getPrevLogIndex());
+ assertEquals("getPrevLogTerm", expected.getPrevLogTerm(), actual.getPrevLogTerm());
+
+ assertEquals("getEntries size", expected.getEntries().size(), actual.getEntries().size());
+ Iterator<ReplicatedLogEntry> iter = expected.getEntries().iterator();
+ for(ReplicatedLogEntry e: actual.getEntries()) {
+ verifyReplicatedLogEntry(iter.next(), e);
+ }
+ }
+
+ private void verifyReplicatedLogEntry(ReplicatedLogEntry expected, ReplicatedLogEntry actual) {
+ assertEquals("getIndex", expected.getIndex(), actual.getIndex());
+ assertEquals("getTerm", expected.getTerm(), actual.getTerm());
+ assertEquals("getData", expected.getData().toString(), actual.getData().toString());
+ }
+}
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
public class ValueSerializer {
+ private static final String NULL_VALUE = "";
+
public static void serialize(NormalizedNodeMessages.Node.Builder builder,
QNameSerializationContext context, Object value) {
builder.setIntValueType(ValueType.getSerializableType(value).ordinal());
}
}
}
- } else if(value instanceof byte[]){
+ } else if(value instanceof byte[]) {
builder.setBytesValue(ByteString.copyFrom((byte[]) value));
+ } else if(value == null){
+ builder.setValue(NULL_VALUE);
} else {
builder.setValue(value.toString());
}
}
} else if(value instanceof byte[]){
builder.setBytesValue(ByteString.copyFrom((byte[]) value));
+ } else if(value == null){
+ builder.setValue(NULL_VALUE);
} else {
builder.setValue(value.toString());
}
package org.opendaylight.controller.cluster.datastore.node.utils.serialization;
-import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMap.Builder;
import java.math.BigDecimal;
Object deserialize(final String str) {
throw new UnsupportedOperationException("Should have been caught by caller");
}
+ },
+ NULL_TYPE {
+ @Override
+ Object deserialize(final String str) {
+ return null;
+ }
};
private static final Map<Class<?>, ValueType> TYPES;
abstract Object deserialize(String str);
public static final ValueType getSerializableType(Object node) {
- Preconditions.checkNotNull(node, "node should not be null");
+ if(node == null){
+ return NULL_TYPE;
+ }
final ValueType type = TYPES.get(node.getClass());
if (type != null) {
return bytes;
case ValueTypes.YANG_IDENTIFIER_TYPE :
- return readYangInstanceIdentifier();
+ return readYangInstanceIdentifier();
default :
return null;
import java.util.Set;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeStreamWriter;
+import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private final Map<String, Integer> stringCodeMap = new HashMap<>();
+ private NormalizedNodeWriter normalizedNodeWriter;
+
public NormalizedNodeOutputStreamWriter(OutputStream stream) throws IOException {
Preconditions.checkNotNull(stream);
output = new DataOutputStream(stream);
this.output = Preconditions.checkNotNull(output);
}
+ private NormalizedNodeWriter normalizedNodeWriter() {
+ if(normalizedNodeWriter == null) {
+ normalizedNodeWriter = NormalizedNodeWriter.forStreamWriter(this);
+ }
+
+ return normalizedNodeWriter;
+ }
+
+ public void writeNormalizedNode(NormalizedNode<?, ?> node) throws IOException {
+ normalizedNodeWriter().write(node);
+ }
+
@Override
public void leafNode(YangInstanceIdentifier.NodeIdentifier name, Object value) throws IOException, IllegalArgumentException {
Preconditions.checkNotNull(name, "Node identifier should not be null");
case ValueTypes.YANG_IDENTIFIER_TYPE:
writeYangInstanceIdentifier((YangInstanceIdentifier) value);
break;
+ case ValueTypes.NULL_TYPE :
+ break;
default:
output.writeUTF(value.toString());
break;
package org.opendaylight.controller.cluster.datastore.node.utils.stream;
-import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMap.Builder;
import java.math.BigDecimal;
public static final byte BIG_INTEGER_TYPE = 10;
public static final byte BIG_DECIMAL_TYPE = 11;
public static final byte BINARY_TYPE = 12;
+ public static final byte NULL_TYPE = 13;
private static final Map<Class<?>, Byte> TYPES;
}
public static final byte getSerializableType(Object node) {
- Preconditions.checkNotNull(node, "node should not be null");
+ if(node == null){
+ return NULL_TYPE;
+ }
final Byte type = TYPES.get(node.getClass());
if (type != null) {
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Deprecated
public class CompositeModificationByteStringPayload extends Payload implements
Serializable {
private static final long serialVersionUID = 1L;
return null;
}
+ @Override
public int size(){
return byteString.size();
}
import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
+@Deprecated
public class CompositeModificationPayload extends Payload implements
Serializable {
- private static final long serialVersionUID = 1L;
private final PersistentMessages.CompositeModification modification;
return this.modification;
}
+ @Override
public int size(){
return this.modification.getSerializedSize();
}
*
*/
public abstract class Payload {
- private String clientPayloadClassName;
public String getClientPayloadClassName() {
return this.getClass().getName();
}
- public void setClientPayloadClassName(String clientPayloadClassName) {
- this.clientPayloadClassName = clientPayloadClassName;
- }
-
/**
* Encode the payload data as a protocol buffer extension.
* <p>
* @param <T>
* @return Map of <GeneratedMessage.GeneratedExtension, T>
*/
+ @Deprecated
public abstract <T extends Object> Map<GeneratedMessage.GeneratedExtension, T> encode();
/**
* @param payload The payload in protocol buffer format
* @return
*/
+ @Deprecated
public abstract Payload decode(
AppendEntriesMessages.AppendEntries.ReplicatedLogEntry.Payload payload);
public abstract int size();
-
-
-
}
}
+ @Test
+ public void testSerializeNull(){
+ NormalizedNodeMessages.Node.Builder builder = NormalizedNodeMessages.Node.newBuilder();
+ Object none = null;
+ ValueSerializer.serialize(builder, mock(QNameSerializationContext.class),none);
+
+ assertEquals(ValueType.NULL_TYPE.ordinal(), builder.getIntValueType());
+ assertEquals("", builder.getValue());
+
+ NormalizedNodeMessages.PathArgumentAttribute.Builder builder1 = NormalizedNodeMessages.PathArgumentAttribute.newBuilder();
+
+ ValueSerializer.serialize(builder1, mock(QNameSerializationContext.class),none);
+
+ assertEquals(ValueType.NULL_TYPE.ordinal(), builder1.getType());
+ assertEquals("", builder.getValue());
+
+ }
+
@Test
public void testDeSerializeShort(){
}
+ @Test
+ public void testDeSerializeNullType(){
+ NormalizedNodeMessages.Node.Builder nodeBuilder = NormalizedNodeMessages.Node.newBuilder();
+ nodeBuilder.setIntValueType(ValueType.NULL_TYPE.ordinal());
+ nodeBuilder.setValue("");
+
+ Object o = ValueSerializer
+ .deSerialize(mock(QNameDeSerializationContext.class),
+ nodeBuilder.build());
+
+ assertEquals(null, o);
+
+ NormalizedNodeMessages.PathArgumentAttribute.Builder argumentBuilder
+ = NormalizedNodeMessages.PathArgumentAttribute.newBuilder();
+
+ argumentBuilder.setType(ValueType.NULL_TYPE.ordinal());
+ argumentBuilder.setValue("");
+
+ o = ValueSerializer
+ .deSerialize(mock(QNameDeSerializationContext.class),
+ argumentBuilder.build());
+
+ assertEquals(null, o);
+
+ }
+
}
package org.opendaylight.controller.cluster.datastore.node.utils.serialization;
-import org.junit.Test;
-
import static org.junit.Assert.assertEquals;
+import org.junit.Test;
public class ValueTypeTest {
ValueType serializableType = ValueType.getSerializableType(b);
assertEquals(ValueType.BINARY_TYPE, serializableType);
}
+
+ @Test
+ public void testNullType(){
+ ValueType serializableType = ValueType.getSerializableType(null);
+ assertEquals(ValueType.NULL_TYPE, serializableType);
+
+ assertEquals(null, ValueType.NULL_TYPE.deserialize(""));
+ }
}
\ No newline at end of file
new YangInstanceIdentifier.NodeWithValue(TestModel.BINARY_LEAF_LIST_QNAME, bytes2)).
withValue(bytes2).build();
+ LeafSetEntryNode<Object> entry3 = ImmutableLeafSetEntryNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeWithValue(TestModel.BINARY_LEAF_LIST_QNAME, null)).
+ withValue(null).build();
+
+
return TestModel.createBaseTestContainerBuilder().
withChild(ImmutableLeafSetNodeBuilder.create().withNodeIdentifier(
new YangInstanceIdentifier.NodeIdentifier(TestModel.BINARY_LEAF_LIST_QNAME)).
- withChild(entry1).withChild(entry2).build()).
+ withChild(entry1).withChild(entry2).withChild(entry3).build()).
withChild(ImmutableNodes.leafNode(TestModel.SOME_BINARY_DATA_QNAME, new byte[]{1,2,3,4})).
withChild(Builders.orderedMapBuilder().
withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(TestModel.ORDERED_LIST_QNAME)).
import akka.actor.ActorSelection;
import com.google.common.base.Optional;
-import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
import java.util.concurrent.Semaphore;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
}
@Override
- public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
- YangInstanceIdentifier path) {
+ public void readData(final YangInstanceIdentifier path, SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture) {
LOG.debug("Tx {} readData called path = {}", identifier, path);
operationLimiter.release();
- return Futures.immediateFailedCheckedFuture(new ReadFailedException(
- "Error reading data for path " + path, failure));
+ proxyFuture.setException(new ReadFailedException("Error reading data for path " + path, failure));
}
@Override
- public CheckedFuture<Boolean, ReadFailedException> dataExists(
- YangInstanceIdentifier path) {
+ public void dataExists(YangInstanceIdentifier path, SettableFuture<Boolean> proxyFuture) {
LOG.debug("Tx {} dataExists called path = {}", identifier, path);
operationLimiter.release();
- return Futures.immediateFailedCheckedFuture(new ReadFailedException(
- "Error checking exists for path " + path, failure));
+ proxyFuture.setException(new ReadFailedException("Error checking exists for path " + path, failure));
}
-}
\ No newline at end of file
+}
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Cancellable;
-import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
+import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateSnapshot;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
-import org.opendaylight.controller.cluster.datastore.messages.ReadData;
-import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
+import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
-import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
import org.opendaylight.controller.cluster.raft.RaftActor;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
-import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
-import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
*/
public class Shard extends RaftActor {
+ private static final YangInstanceIdentifier DATASTORE_ROOT = YangInstanceIdentifier.builder().build();
+
private static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck";
@VisibleForTesting
private SchemaContext schemaContext;
- private ActorRef createSnapshotTransaction;
-
private int createSnapshotTransactionCounter;
private final ShardCommitCoordinator commitCoordinator;
LOG.debug("onReceiveCommand: Received message {} from {}", message, getSender());
}
- if(message.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
- handleReadDataReply(message);
- } else if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
+ if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
handleCreateTransaction(message);
} else if(message instanceof ForwardedReadyTransaction) {
handleForwardedReadyTransaction((ForwardedReadyTransaction)message);
applyModificationToState(getSender(), transactionID, cohortEntry.getModification());
} else {
Shard.this.persistData(getSender(), transactionID,
- new CompositeModificationByteStringPayload(cohortEntry.getModification().toSerializable()));
+ new ModificationPayload(cohortEntry.getModification()));
}
- } catch (InterruptedException | ExecutionException e) {
+ } catch (InterruptedException | ExecutionException | IOException e) {
LOG.error(e, "An exception occurred while preCommitting transaction {}",
cohortEntry.getTransactionID());
shardMBean.incrementFailedTransactionsCount();
}
}
- private void handleReadDataReply(final Object message) {
- // This must be for install snapshot. Don't want to open this up and trigger
- // deSerialization
-
- self().tell(new CaptureSnapshotReply(ReadDataReply.fromSerializableAsByteString(message)),
- self());
-
- createSnapshotTransaction = null;
-
- // Send a PoisonPill instead of sending close transaction because we do not really need
- // a response
- getSender().tell(PoisonPill.getInstance(), self());
- }
-
private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
DOMStoreTransactionChain chain =
transactionChains.remove(closeTransactionChain.getTransactionChainId());
@Override
protected void appendRecoveredLogEntry(final Payload data) {
- if (data instanceof CompositeModificationPayload) {
+ if(data instanceof ModificationPayload) {
+ try {
+ currentLogRecoveryBatch.add(((ModificationPayload) data).getModification());
+ } catch (ClassNotFoundException | IOException e) {
+ LOG.error(e, "Error extracting ModificationPayload");
+ }
+ } else if (data instanceof CompositeModificationPayload) {
currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification());
} else if (data instanceof CompositeModificationByteStringPayload) {
currentLogRecoveryBatch.add(((CompositeModificationByteStringPayload) data).getModification());
}
@Override
- protected void applyRecoverySnapshot(final ByteString snapshot) {
+ protected void applyRecoverySnapshot(final byte[] snapshotBytes) {
if(recoveryCoordinator == null) {
recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
}
- recoveryCoordinator.submit(snapshot, store.newWriteOnlyTransaction());
+ recoveryCoordinator.submit(snapshotBytes, store.newWriteOnlyTransaction());
if(LOG.isDebugEnabled()) {
LOG.debug("{} : submitted recovery sbapshot", persistenceId());
@Override
protected void applyState(final ActorRef clientActor, final String identifier, final Object data) {
- if (data instanceof CompositeModificationPayload) {
+ if(data instanceof ModificationPayload) {
+ try {
+ applyModificationToState(clientActor, identifier, ((ModificationPayload) data).getModification());
+ } catch (ClassNotFoundException | IOException e) {
+ LOG.error(e, "Error extracting ModificationPayload");
+ }
+ }
+ else if (data instanceof CompositeModificationPayload) {
Object modification = ((CompositeModificationPayload) data).getModification();
applyModificationToState(clientActor, identifier, modification);
Object modification = ((CompositeModificationByteStringPayload) data).getModification();
applyModificationToState(clientActor, identifier, modification);
-
} else {
LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}",
data, data.getClass().getClassLoader(),
} else if(clientActor == null) {
// There's no clientActor to which to send a commit reply so we must be applying
// replicated state from the leader.
- commitWithNewTransaction(MutableCompositeModification.fromSerializable(
- modification, schemaContext));
+ commitWithNewTransaction(MutableCompositeModification.fromSerializable(modification));
} else {
// This must be the OK to commit after replication consensus.
finishCommit(clientActor, identifier);
@Override
protected void createSnapshot() {
- if (createSnapshotTransaction == null) {
+ // Create a transaction actor. We are really going to treat the transaction as a worker
+ // so that this actor does not get block building the snapshot. THe transaction actor will
+ // after processing the CreateSnapshot message.
- // Create a transaction. We are really going to treat the transaction as a worker
- // so that this actor does not get block building the snapshot
- createSnapshotTransaction = createTransaction(
+ ActorRef createSnapshotTransaction = createTransaction(
TransactionProxy.TransactionType.READ_ONLY.ordinal(),
"createSnapshot" + ++createSnapshotTransactionCounter, "",
DataStoreVersions.CURRENT_VERSION);
- createSnapshotTransaction.tell(
- new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
-
- }
+ createSnapshotTransaction.tell(CreateSnapshot.INSTANCE, self());
}
@VisibleForTesting
@Override
- protected void applySnapshot(final ByteString snapshot) {
+ protected void applySnapshot(final byte[] snapshotBytes) {
// Since this will be done only on Recovery or when this actor is a Follower
// we can safely commit everything in here. We not need to worry about event notifications
// as they would have already been disabled on the follower
LOG.info("Applying snapshot");
try {
DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
- NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
- NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext)
- .decode(serializedNode);
+
+ NormalizedNode<?, ?> node = SerializationUtils.deserializeNormalizedNode(snapshotBytes);
// delete everything first
- transaction.delete(YangInstanceIdentifier.builder().build());
+ transaction.delete(DATASTORE_ROOT);
// Add everything from the remote node back
- transaction.write(YangInstanceIdentifier.builder().build(), node);
+ transaction.write(DATASTORE_ROOT, node);
syncCommitTransaction(transaction);
- } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) {
+ } catch (InterruptedException | ExecutionException e) {
LOG.error(e, "An exception occurred when applying snapshot");
} finally {
LOG.info("Done applying snapshot");
package org.opendaylight.controller.cluster.datastore;
import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
+import org.opendaylight.controller.cluster.datastore.messages.CreateSnapshot;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.controller.cluster.datastore.messages.ReadData;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
+import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
/**
* Date: 8/6/14
*/
public class ShardReadTransaction extends ShardTransaction {
+ private static final YangInstanceIdentifier DATASTORE_ROOT = YangInstanceIdentifier.builder().build();
+
private final DOMStoreReadTransaction transaction;
public ShardReadTransaction(DOMStoreReadTransaction transaction, ActorRef shardActor,
} else if (message instanceof DataExists) {
dataExists(transaction, (DataExists) message, !SERIALIZED_REPLY);
-
+ } else if (message instanceof CreateSnapshot) {
+ createSnapshot();
} else if(ReadData.SERIALIZABLE_CLASS.equals(message.getClass())) {
readData(transaction, ReadData.fromSerializable(message), SERIALIZED_REPLY);
}
}
+ private void createSnapshot() {
+
+ // This is a special message sent by the shard to send back a serialized snapshot of the whole
+ // data store tree. This transaction was created for that purpose only so we can
+ // self-destruct after sending the reply.
+
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(DATASTORE_ROOT);
+
+ Futures.addCallback(future, new FutureCallback<Optional<NormalizedNode<?, ?>>>() {
+ @Override
+ public void onSuccess(Optional<NormalizedNode<?, ?>> result) {
+ byte[] serialized = SerializationUtils.serializeNormalizedNode(result.get());
+ sender.tell(new CaptureSnapshotReply(serialized), self);
+
+ self.tell(PoisonPill.getInstance(), self);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ sender.tell(new akka.actor.Status.Failure(t), self);
+
+ self.tell(PoisonPill.getInstance(), self);
+ }
+ });
+ }
+
@Override
protected DOMStoreTransaction getDOMStoreTransaction() {
return transaction;
*/
package org.opendaylight.controller.cluster.datastore;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
-import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
-import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
-
/**
* Coordinates persistence recovery of journal log entries and snapshots for a shard. Each snapshot
* and journal log entry batch are de-serialized and applied to their own write transaction
/**
* Submits a snapshot.
*
- * @param snapshot the serialized snapshot
+ * @param snapshotBytes the serialized snapshot
* @param resultingTx the write Tx to which to apply the entries
*/
- void submit(ByteString snapshot, DOMStoreWriteTransaction resultingTx) {
- SnapshotRecoveryTask task = new SnapshotRecoveryTask(snapshot, resultingTx);
+ void submit(byte[] snapshotBytes, DOMStoreWriteTransaction resultingTx) {
+ SnapshotRecoveryTask task = new SnapshotRecoveryTask(snapshotBytes, resultingTx);
resultingTxList.add(resultingTx);
executor.execute(task);
}
public void run() {
for(int i = 0; i < logEntries.size(); i++) {
MutableCompositeModification.fromSerializable(
- logEntries.get(i), schemaContext).apply(resultingTx);
+ logEntries.get(i)).apply(resultingTx);
// Null out to GC quicker.
logEntries.set(i, null);
}
private class SnapshotRecoveryTask extends ShardRecoveryTask {
- private final ByteString snapshot;
+ private final byte[] snapshotBytes;
- SnapshotRecoveryTask(ByteString snapshot, DOMStoreWriteTransaction resultingTx) {
+ SnapshotRecoveryTask(byte[] snapshotBytes, DOMStoreWriteTransaction resultingTx) {
super(resultingTx);
- this.snapshot = snapshot;
+ this.snapshotBytes = snapshotBytes;
}
@Override
public void run() {
- try {
- NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
- NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext).decode(
- serializedNode);
-
- // delete everything first
- resultingTx.delete(YangInstanceIdentifier.builder().build());
-
- // Add everything from the remote node back
- resultingTx.write(YangInstanceIdentifier.builder().build(), node);
- } catch (InvalidProtocolBufferException e) {
- LOG.error("Error deserializing snapshot", e);
- }
+ NormalizedNode<?, ?> node = SerializationUtils.deserializeNormalizedNode(snapshotBytes);
+
+ // delete everything first
+ resultingTx.delete(YangInstanceIdentifier.builder().build());
+
+ // Add everything from the remote node back
+ resultingTx.write(YangInstanceIdentifier.builder().build(), node);
}
}
}
import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
-import org.opendaylight.controller.cluster.datastore.modification.ImmutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
} else if (message instanceof GetCompositedModification) {
// This is here for testing only
- getSender().tell(new GetCompositeModificationReply(
- new ImmutableCompositeModification(modification)), getSelf());
+ getSender().tell(new GetCompositeModificationReply(modification), getSelf());
} else {
super.handleReceive(message);
}
LOG.debug("writeData at path : {}", message.getPath());
modification.addModification(
- new WriteModification(message.getPath(), message.getData(), getSchemaContext()));
+ new WriteModification(message.getPath(), message.getData()));
try {
transaction.write(message.getPath(), message.getData());
WriteDataReply writeDataReply = WriteDataReply.INSTANCE;
LOG.debug("mergeData at path : {}", message.getPath());
modification.addModification(
- new MergeModification(message.getPath(), message.getData(), getSchemaContext()));
+ new MergeModification(message.getPath(), message.getData()));
try {
transaction.merge(message.getPath(), message.getData());
import akka.actor.ActorSelection;
import com.google.common.base.Optional;
-import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.SettableFuture;
import java.util.List;
-import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import scala.concurrent.Future;
void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
- CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
- final YangInstanceIdentifier path);
+ void readData(final YangInstanceIdentifier path, SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture);
- CheckedFuture<Boolean, ReadFailedException> dataExists(YangInstanceIdentifier path);
+ void dataExists(YangInstanceIdentifier path, SettableFuture<Boolean> proxyFuture);
List<Future<Object>> getRecordedOperationFutures();
-}
\ No newline at end of file
+}
import akka.dispatch.OnComplete;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.List;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.WriteData;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
-import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
}
@Override
- public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
- final YangInstanceIdentifier path) {
+ public void readData(
+ final YangInstanceIdentifier path,final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture ) {
LOG.debug("Tx {} readData called path = {}", identifier, path);
- final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture = SettableFuture.create();
-
// If there were any previous recorded put/merge/delete operation reply Futures then we
// must wait for them to successfully complete. This is necessary to honor the read
// uncommitted semantics of the public API contract. If any one fails then fail the read.
combinedFutures.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
}
- return MappingCheckedFuture.create(returnFuture, ReadFailedException.MAPPER);
}
private void finishReadData(final YangInstanceIdentifier path,
}
@Override
- public CheckedFuture<Boolean, ReadFailedException> dataExists(
- final YangInstanceIdentifier path) {
+ public void dataExists(final YangInstanceIdentifier path, final SettableFuture<Boolean> returnFuture) {
LOG.debug("Tx {} dataExists called path = {}", identifier, path);
- final SettableFuture<Boolean> returnFuture = SettableFuture.create();
-
// If there were any previous recorded put/merge/delete operation reply Futures then we
// must wait for them to successfully complete. This is necessary to honor the read
// uncommitted semantics of the public API contract. If any one fails then fail this
combinedFutures.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
}
-
- return MappingCheckedFuture.create(returnFuture, ReadFailedException.MAPPER);
}
private void finishDataExists(final YangInstanceIdentifier path,
future.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
}
-}
\ No newline at end of file
+}
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import java.util.ArrayList;
import java.util.Collection;
throttleOperation();
+ final SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture = SettableFuture.create();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- return txFutureCallback.enqueueReadOperation(new ReadOperation<Optional<NormalizedNode<?, ?>>>() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
- public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> invoke(
- TransactionContext transactionContext) {
- return transactionContext.readData(path);
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.readData(path, proxyFuture);
}
});
+
+ return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
}
@Override
throttleOperation();
+ final SettableFuture<Boolean> proxyFuture = SettableFuture.create();
+
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- return txFutureCallback.enqueueReadOperation(new ReadOperation<Boolean>() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
- public CheckedFuture<Boolean, ReadFailedException> invoke(TransactionContext transactionContext) {
- return transactionContext.dataExists(path);
+ public void invoke(TransactionContext transactionContext) {
+ transactionContext.dataExists(path, proxyFuture);
}
});
- }
+ return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
+ }
private void checkModificationState() {
Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
public void invoke(TransactionContext transactionContext) {
transactionContext.writeData(path, data);
throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
public void invoke(TransactionContext transactionContext) {
transactionContext.mergeData(path, data);
throttleOperation();
TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
- txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
public void invoke(TransactionContext transactionContext) {
transactionContext.deleteData(path);
LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
txFutureCallback.getShardName(), transactionChainId);
- Future<ActorSelection> future = txFutureCallback.enqueueFutureOperation(new FutureOperation<ActorSelection>() {
- @Override
- public Future<ActorSelection> invoke(TransactionContext transactionContext) {
- return transactionContext.readyTransaction();
- }
- });
+ final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
+ final Future<ActorSelection> future;
+ if (transactionContext != null) {
+ // avoid the creation of a promise and a TransactionOperation
+ future = transactionContext.readyTransaction();
+ } else {
+ final Promise<ActorSelection> promise = akka.dispatch.Futures.promise();
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
+ @Override
+ public void invoke(TransactionContext transactionContext) {
+ promise.completeWith(transactionContext.readyTransaction());
+ }
+ });
+ future = promise.future();
+ }
cohortFutures.add(future);
}
@Override
public void close() {
for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
- txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
+ txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
@Override
public void invoke(TransactionContext transactionContext) {
transactionContext.closeTransaction();
void invoke(TransactionContext transactionContext);
}
- /**
- * This interface returns a Guava Future
- */
- private static interface ReadOperation<T> {
- CheckedFuture<T, ReadFailedException> invoke(TransactionContext transactionContext);
- }
-
- /**
- * This interface returns a Scala Future
- */
- private static interface FutureOperation<T> {
- Future<T> invoke(TransactionContext transactionContext);
- }
-
/**
* Implements a Future OnComplete callback for a CreateTransaction message. This class handles
* retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
}
}
-
- <T> Future<T> enqueueFutureOperation(final FutureOperation<T> op) {
-
- Future<T> future;
-
- if (transactionContext != null) {
- future = op.invoke(transactionContext);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- final Promise<T> promise = akka.dispatch.Futures.promise();
- addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- promise.completeWith(op.invoke(transactionContext));
- }
- });
-
- future = promise.future();
- }
-
- return future;
- }
-
- <T> CheckedFuture<T, ReadFailedException> enqueueReadOperation(final ReadOperation<T> op) {
-
- CheckedFuture<T, ReadFailedException> future;
-
- if (transactionContext != null) {
- future = op.invoke(transactionContext);
- } else {
- // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
- // callback to be executed after the Tx is created.
- final SettableFuture<T> proxyFuture = SettableFuture.create();
- addTxOperationOnComplete(new TransactionOperation() {
- @Override
- public void invoke(TransactionContext transactionContext) {
- Futures.addCallback(op.invoke(transactionContext), new FutureCallback<T>() {
- @Override
- public void onSuccess(T data) {
- proxyFuture.set(data);
- }
-
- @Override
- public void onFailure(Throwable t) {
- proxyFuture.setException(t);
- }
- });
- }
- });
-
- future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
- }
-
- return future;
- }
-
- void enqueueModifyOperation(final TransactionOperation op) {
+ void enqueueTransactionOperation(final TransactionOperation op) {
if (transactionContext != null) {
op.invoke(transactionContext);
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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.datastore.messages;
+
+/**
+ * Message sent to a transaction actor to create a snapshot of the data store.
+ *
+ * @author Thomas Pantelis
+ */
+public class CreateSnapshot {
+ // Note: This class does not need to Serializable as it's only sent locally.
+
+ public static final CreateSnapshot INSTANCE = new CreateSnapshot();
+}
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import java.io.Serializable;
-
/**
* Base class to be used for all simple modifications that can be applied to a DOMStoreTransaction
*/
-public abstract class AbstractModification implements Modification,
- Serializable {
+public abstract class AbstractModification implements Modification {
- private static final long serialVersionUID = 1638042650152084457L;
+ private YangInstanceIdentifier path;
- protected final YangInstanceIdentifier path;
+ protected AbstractModification() {
+ }
protected AbstractModification(YangInstanceIdentifier path) {
this.path = path;
}
+ protected void setPath(YangInstanceIdentifier path) {
+ this.path = path;
+ }
+
public YangInstanceIdentifier getPath() {
return path;
}
package org.opendaylight.controller.cluster.datastore.modification;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.util.InstanceIdentifierUtils;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
public class DeleteModification extends AbstractModification {
private static final long serialVersionUID = 1L;
+ public DeleteModification() {
+ }
+
public DeleteModification(YangInstanceIdentifier path) {
super(path);
}
@Override
public void apply(DOMStoreWriteTransaction transaction) {
- transaction.delete(path);
+ transaction.delete(getPath());
}
@Override
+ public byte getType() {
+ return DELETE;
+ }
+
+ @Override
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ in.readShort();
+ setPath(SerializationUtils.deserializePath(in));
+ }
+
+ @Override
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeShort(DataStoreVersions.CURRENT_VERSION);
+ SerializationUtils.serializePath(getPath(), out);
+ }
+
+ @Override
+ @Deprecated
public Object toSerializable() {
return PersistentMessages.Modification.newBuilder().setType(this.getClass().toString())
- .setPath(InstanceIdentifierUtils.toSerializable(this.path)).build();
+ .setPath(InstanceIdentifierUtils.toSerializable(getPath())).build();
}
+ @Deprecated
public static DeleteModification fromSerializable(Object serializable) {
PersistentMessages.Modification o = (PersistentMessages.Modification) serializable;
return new DeleteModification(InstanceIdentifierUtils.fromSerializable(o.getPath()));
}
+
+ public static DeleteModification fromStream(ObjectInput in) throws ClassNotFoundException, IOException {
+ DeleteModification mod = new DeleteModification();
+ mod.readExternal(in);
+ return mod;
+ }
}
+++ /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.datastore.modification;
-
-import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
-
-import java.util.List;
-
-public class ImmutableCompositeModification implements CompositeModification {
-
- private final CompositeModification modification;
-
- public ImmutableCompositeModification(CompositeModification modification) {
- this.modification = modification;
- }
-
- @Override
- public List<Modification> getModifications() {
- return modification.getModifications();
- }
-
- @Override
- public void apply(DOMStoreWriteTransaction transaction) {
- modification.apply(transaction);
- }
-
- @Override public Object toSerializable() {
-
- PersistentMessages.CompositeModification.Builder builder =
- PersistentMessages.CompositeModification.newBuilder();
-
- for (Modification m : modification.getModifications()) {
- builder.addModification(
- (PersistentMessages.Modification) m.toSerializable());
- }
-
- return builder.build();
- }
-}
package org.opendaylight.controller.cluster.datastore.modification;
+import java.io.IOException;
+import java.io.ObjectInput;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec.Decoded;
import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
/**
* MergeModification stores all the parameters required to merge data into the specified path
public class MergeModification extends WriteModification {
private static final long serialVersionUID = 1L;
- public MergeModification(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
- final SchemaContext schemaContext) {
- super(path, data, schemaContext);
+ public MergeModification() {
+ }
+
+ public MergeModification(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+ super(path, data);
}
@Override
public void apply(final DOMStoreWriteTransaction transaction) {
- transaction.merge(path, data);
+ transaction.merge(getPath(), getData());
}
- public static MergeModification fromSerializable(final Object serializable, final SchemaContext schemaContext) {
+ @Override
+ public byte getType() {
+ return MERGE;
+ }
+
+ @Deprecated
+ public static MergeModification fromSerializable(final Object serializable) {
PersistentMessages.Modification o = (PersistentMessages.Modification) serializable;
- Decoded decoded = new NormalizedNodeToNodeCodec(schemaContext).decode(o.getPath(), o.getData());
- return new MergeModification(decoded.getDecodedPath(), decoded.getDecodedNode(), schemaContext);
+ Decoded decoded = new NormalizedNodeToNodeCodec(null).decode(o.getPath(), o.getData());
+ return new MergeModification(decoded.getDecodedPath(), decoded.getDecodedNode());
+ }
+
+ public static MergeModification fromStream(ObjectInput in) throws ClassNotFoundException, IOException {
+ MergeModification mod = new MergeModification();
+ mod.readExternal(in);
+ return mod;
}
}
package org.opendaylight.controller.cluster.datastore.modification;
-import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage;
+import java.io.Externalizable;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
/**
* which can then be applied to a write transaction
* </p>
*/
-public interface Modification extends SerializableMessage {
- /**
- * Apply the modification to the specified transaction
- * @param transaction
- */
- void apply(DOMStoreWriteTransaction transaction);
+public interface Modification extends Externalizable {
+
+ byte COMPOSITE = 1;
+ byte WRITE = 2;
+ byte MERGE = 3;
+ byte DELETE = 4;
+
+ /**
+ * Apply the modification to the specified transaction
+ *
+ * @param transaction
+ */
+ void apply(DOMStoreWriteTransaction transaction);
+
+ byte getType();
+
+ @Deprecated
+ Object toSerializable();
}
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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.datastore.modification;
+
+import com.google.protobuf.GeneratedMessage.GeneratedExtension;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
+import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages.AppendEntries.ReplicatedLogEntry;
+
+/**
+ * Payload implementation for MutableCompositeModification used for persistence and replication.
+ *
+ * @author Thomas Pantelis
+ */
+public class ModificationPayload extends Payload implements Externalizable {
+ private static final long serialVersionUID = 1L;
+
+ private transient byte[] serializedPayload;
+
+ public ModificationPayload() {
+ }
+
+ public ModificationPayload(Modification from) throws IOException {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(bos);
+ out.writeObject(from);
+ out.close();
+ serializedPayload = bos.toByteArray();
+ }
+
+ public Modification getModification() throws IOException, ClassNotFoundException {
+ ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(serializedPayload));
+ Modification to = (Modification) in.readObject();
+ in.close();
+ return to;
+ }
+
+ @Override
+ public int size() {
+ return serializedPayload.length;
+ }
+
+ @Override
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ int size = in.readInt();
+ serializedPayload = new byte[size];
+ in.readFully(serializedPayload);
+ }
+
+ @Override
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(serializedPayload.length);
+ out.write(serializedPayload);
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ @Deprecated
+ public <T> Map<GeneratedExtension, T> encode() {
+ return null;
+ }
+
+ @Override
+ @Deprecated
+ public Payload decode(ReplicatedLogEntry.Payload payload) {
+ return null;
+ }
+}
package org.opendaylight.controller.cluster.datastore.modification;
-import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
+import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
+import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeInputStreamReader;
+import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeOutputStreamWriter;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
+import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
/**
* MutableCompositeModification is just a mutable version of a
* CompositeModification {@link org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification#addModification(Modification)}
*/
-public class MutableCompositeModification
- implements CompositeModification {
+public class MutableCompositeModification implements CompositeModification {
+ private static final long serialVersionUID = 1L;
- private static final long serialVersionUID = 1163377899140186790L;
+ private final List<Modification> modifications;
- private final List<Modification> modifications = new ArrayList<>();
+ public MutableCompositeModification() {
+ modifications = new ArrayList<>();
+ }
@Override
public void apply(DOMStoreWriteTransaction transaction) {
}
}
+ @Override
+ public byte getType() {
+ return COMPOSITE;
+ }
+
/**
* Add a new Modification to the list of Modifications represented by this
* composite
modifications.add(modification);
}
+ @Override
public List<Modification> getModifications() {
- return Collections.unmodifiableList(modifications);
+ return modifications;
+ }
+
+ @Override
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ in.readShort();
+
+ int size = in.readInt();
+
+ if(size > 1) {
+ SerializationUtils.REUSABLE_READER_TL.set(new NormalizedNodeInputStreamReader(in));
+ }
+
+ try {
+ for(int i = 0; i < size; i++) {
+ byte type = in.readByte();
+ switch(type) {
+ case Modification.WRITE:
+ modifications.add(WriteModification.fromStream(in));
+ break;
+
+ case Modification.MERGE:
+ modifications.add(MergeModification.fromStream(in));
+ break;
+
+ case Modification.DELETE:
+ modifications.add(DeleteModification.fromStream(in));
+ break;
+ }
+ }
+ } finally {
+ SerializationUtils.REUSABLE_READER_TL.remove();
+ }
+ }
+
+ @Override
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeShort(DataStoreVersions.CURRENT_VERSION);
+
+ out.writeInt(modifications.size());
+
+ if(modifications.size() > 1) {
+ SerializationUtils.REUSABLE_WRITER_TL.set(new NormalizedNodeOutputStreamWriter(out));
+ }
+
+ try {
+ for(Modification mod: modifications) {
+ out.writeByte(mod.getType());
+ mod.writeExternal(out);
+ }
+ } finally {
+ SerializationUtils.REUSABLE_WRITER_TL.remove();
+ }
}
- @Override public Object toSerializable() {
+ @Override
+ @Deprecated
+ public Object toSerializable() {
PersistentMessages.CompositeModification.Builder builder =
- PersistentMessages.CompositeModification.newBuilder();
+ PersistentMessages.CompositeModification.newBuilder();
builder.setTimeStamp(System.nanoTime());
for (Modification m : modifications) {
builder.addModification(
- (PersistentMessages.Modification) m.toSerializable());
+ (PersistentMessages.Modification) m.toSerializable());
}
return builder.build();
}
- public static MutableCompositeModification fromSerializable(Object serializable, SchemaContext schemaContext){
+ public static MutableCompositeModification fromSerializable(Object serializable) {
+ if(serializable instanceof MutableCompositeModification) {
+ return (MutableCompositeModification)serializable;
+ } else {
+ return fromLegacySerializable(serializable);
+ }
+ }
+
+ private static MutableCompositeModification fromLegacySerializable(Object serializable) {
PersistentMessages.CompositeModification o = (PersistentMessages.CompositeModification) serializable;
MutableCompositeModification compositeModification = new MutableCompositeModification();
if(m.getType().equals(DeleteModification.class.toString())){
compositeModification.addModification(DeleteModification.fromSerializable(m));
} else if(m.getType().equals(WriteModification.class.toString())){
- compositeModification.addModification(WriteModification.fromSerializable(m, schemaContext));
+ compositeModification.addModification(WriteModification.fromSerializable(m));
} else if(m.getType().equals(MergeModification.class.toString())){
- compositeModification.addModification(MergeModification.fromSerializable(m, schemaContext));
+ compositeModification.addModification(MergeModification.fromSerializable(m));
}
}
package org.opendaylight.controller.cluster.datastore.modification;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec.Decoded;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec.Encoded;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils.Applier;
import org.opendaylight.controller.protobuff.messages.persistent.PersistentMessages;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
/**
* WriteModification stores all the parameters required to write data to the specified path
*/
public class WriteModification extends AbstractModification {
private static final long serialVersionUID = 1L;
- protected final NormalizedNode<?, ?> data;
- private final SchemaContext schemaContext;
- public WriteModification(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data, final SchemaContext schemaContext) {
+ private NormalizedNode<?, ?> data;
+
+ public WriteModification() {
+ }
+
+ public WriteModification(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
super(path);
this.data = data;
- this.schemaContext = schemaContext;
}
@Override
public void apply(final DOMStoreWriteTransaction transaction) {
- transaction.write(path, data);
+ transaction.write(getPath(), data);
}
public NormalizedNode<?, ?> getData() {
}
@Override
- public Object toSerializable() {
- Encoded encoded = new NormalizedNodeToNodeCodec(schemaContext).encode(path, data);
+ public byte getType() {
+ return WRITE;
+ }
- return PersistentMessages.Modification.newBuilder()
- .setType(this.getClass().toString())
- .setPath(encoded.getEncodedPath())
- .setData(encoded.getEncodedNode().getNormalizedNode())
- .build();
+ @Override
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ in.readShort(); // version
+
+ SerializationUtils.deserializePathAndNode(in, this, APPLIER);
}
- public static WriteModification fromSerializable(final Object serializable, final SchemaContext schemaContext) {
+ @Override
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeShort(DataStoreVersions.CURRENT_VERSION);
+ SerializationUtils.serializePathAndNode(getPath(), data, out);
+ }
+
+ @Override
+ @Deprecated
+ public Object toSerializable() {
+ Encoded encoded = new NormalizedNodeToNodeCodec(null).encode(getPath(), getData());
+ return PersistentMessages.Modification.newBuilder().setType(this.getClass().toString())
+ .setPath(encoded.getEncodedPath()).setData(encoded.getEncodedNode()
+ .getNormalizedNode()).build();
+ }
+
+ @Deprecated
+ public static WriteModification fromSerializable(final Object serializable) {
PersistentMessages.Modification o = (PersistentMessages.Modification) serializable;
- Decoded decoded = new NormalizedNodeToNodeCodec(schemaContext).decode(o.getPath(), o.getData());
- return new WriteModification(decoded.getDecodedPath(), decoded.getDecodedNode(), schemaContext);
+ Decoded decoded = new NormalizedNodeToNodeCodec(null).decode(o.getPath(), o.getData());
+ return new WriteModification(decoded.getDecodedPath(), decoded.getDecodedNode());
+ }
+
+ public static WriteModification fromStream(ObjectInput in) throws ClassNotFoundException, IOException {
+ WriteModification mod = new WriteModification();
+ mod.readExternal(in);
+ return mod;
}
+
+ private static final Applier<WriteModification> APPLIER = new Applier<WriteModification>() {
+ @Override
+ public void apply(WriteModification instance, YangInstanceIdentifier path,
+ NormalizedNode<?, ?> node) {
+ instance.setPath(path);
+ instance.data = node;
+ }
+ };
}
package org.opendaylight.controller.cluster.datastore.utils;
import com.google.common.base.Preconditions;
+import com.google.protobuf.InvalidProtocolBufferException;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.DataInput;
+import java.io.DataInputStream;
import java.io.DataOutput;
+import java.io.DataOutputStream;
import java.io.IOException;
+import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeInputStreamReader;
import org.opendaylight.controller.cluster.datastore.node.utils.stream.NormalizedNodeOutputStreamWriter;
+import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.data.api.schema.stream.NormalizedNodeWriter;
/**
* Provides various utility methods for serialization and de-serialization.
* @author Thomas Pantelis
*/
public final class SerializationUtils {
+ public static ThreadLocal<NormalizedNodeOutputStreamWriter> REUSABLE_WRITER_TL = new ThreadLocal<>();
+ public static ThreadLocal<NormalizedNodeInputStreamReader> REUSABLE_READER_TL = new ThreadLocal<>();
public static interface Applier<T> {
void apply(T instance, YangInstanceIdentifier path, NormalizedNode<?, ?> node);
}
+ private static NormalizedNodeOutputStreamWriter streamWriter(DataOutput out) throws IOException {
+ NormalizedNodeOutputStreamWriter streamWriter = REUSABLE_WRITER_TL.get();
+ if(streamWriter == null) {
+ streamWriter = new NormalizedNodeOutputStreamWriter(out);
+ }
+
+ return streamWriter;
+ }
+
+ private static NormalizedNodeInputStreamReader streamReader(DataInput in) throws IOException {
+ NormalizedNodeInputStreamReader streamWriter = REUSABLE_READER_TL.get();
+ if(streamWriter == null) {
+ streamWriter = new NormalizedNodeInputStreamReader(in);
+ }
+
+ return streamWriter;
+ }
+
public static void serializePathAndNode(YangInstanceIdentifier path, NormalizedNode<?, ?> node,
DataOutput out) {
Preconditions.checkNotNull(path);
Preconditions.checkNotNull(node);
try {
- NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out);
- NormalizedNodeWriter.forStreamWriter(streamWriter).write(node);
+ NormalizedNodeOutputStreamWriter streamWriter = streamWriter(out);
+ streamWriter.writeNormalizedNode(node);
streamWriter.writeYangInstanceIdentifier(path);
} catch (IOException e) {
throw new IllegalArgumentException(String.format("Error serializing path %s and Node %s",
public static <T> void deserializePathAndNode(DataInput in, T instance, Applier<T> applier) {
try {
- NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in);
+ NormalizedNodeInputStreamReader streamReader = streamReader(in);
NormalizedNode<?, ?> node = streamReader.readNormalizedNode();
YangInstanceIdentifier path = streamReader.readYangInstanceIdentifier();
applier.apply(instance, path, node);
try {
out.writeBoolean(node != null);
if(node != null) {
- NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out);
- NormalizedNodeWriter.forStreamWriter(streamWriter).write(node);
+ NormalizedNodeOutputStreamWriter streamWriter = streamWriter(out);
+ streamWriter.writeNormalizedNode(node);
}
} catch (IOException e) {
throw new IllegalArgumentException(String.format("Error serializing NormalizedNode %s",
try {
boolean present = in.readBoolean();
if(present) {
- NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in);
+ NormalizedNodeInputStreamReader streamReader = streamReader(in);
return streamReader.readNormalizedNode();
}
} catch (IOException e) {
return null;
}
+ public static NormalizedNode<?, ?> deserializeNormalizedNode(byte [] bytes) {
+ NormalizedNode<?, ?> node = null;
+ try {
+ node = deserializeNormalizedNode(new DataInputStream(new ByteArrayInputStream(bytes)));
+ } catch(Exception e) {
+ }
+
+ if(node == null) {
+ // Must be from legacy protobuf serialization - try that.
+ try {
+ NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(bytes);
+ node = new NormalizedNodeToNodeCodec(null).decode(serializedNode);
+ } catch (InvalidProtocolBufferException e) {
+ throw new IllegalArgumentException("Error deserializing NormalizedNode", e);
+ }
+ }
+
+ return node;
+ }
+
+ public static byte [] serializeNormalizedNode(NormalizedNode<?, ?> node) {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ serializeNormalizedNode(node, new DataOutputStream(bos));
+ return bos.toByteArray();
+ }
+
public static void serializePath(YangInstanceIdentifier path, DataOutput out) {
Preconditions.checkNotNull(path);
try {
- NormalizedNodeOutputStreamWriter streamWriter = new NormalizedNodeOutputStreamWriter(out);
+ NormalizedNodeOutputStreamWriter streamWriter = streamWriter(out);
streamWriter.writeYangInstanceIdentifier(path);
} catch (IOException e) {
- throw new IllegalArgumentException(String.format("Error serializing path {}", path), e);
+ throw new IllegalArgumentException(String.format("Error serializing path %s", path), e);
}
}
public static YangInstanceIdentifier deserializePath(DataInput in) {
try {
- NormalizedNodeInputStreamReader streamReader = new NormalizedNodeInputStreamReader(in);
+ NormalizedNodeInputStreamReader streamReader = streamReader(in);
return streamReader.readYangInstanceIdentifier();
} catch (IOException e) {
throw new IllegalArgumentException("Error deserializing path", e);
import java.util.List;
import org.apache.commons.lang.SerializationUtils;
import org.junit.Test;
-import org.opendaylight.controller.cluster.datastore.modification.Modification;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+@Deprecated
public class CompositeModificationByteStringPayloadTest {
private static final SchemaContext SCHEMA_CONTEXT = TestModel.createTestContext();
public void testSerialization(){
WriteModification writeModification =
new WriteModification(TestModel.TEST_PATH, ImmutableNodes
- .containerNode(TestModel.TEST_QNAME),
- TestModel.createTestContext());
+ .containerNode(TestModel.TEST_QNAME));
MutableCompositeModification compositeModification =
new MutableCompositeModification();
public void testAppendEntries(){
List<ReplicatedLogEntry> entries = new ArrayList<>();
- CompositeModificationByteStringPayload payload = newByteStringPayload(
- new WriteModification(TestModel.OUTER_LIST_PATH,
- ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
- SCHEMA_CONTEXT));
+ WriteModification writeModification = new WriteModification(TestModel.OUTER_LIST_PATH,
+ ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
- payload.clearModificationReference();
-
- entries.add(new ReplicatedLogImplEntry(0, 1, payload));
+ MutableCompositeModification compositeModification = new MutableCompositeModification();
+ compositeModification.addModification(writeModification);
- assertNotNull(new AppendEntries(10, "foobar", 10, 10, entries, 10).toSerializable());
- }
-
+ CompositeModificationByteStringPayload payload =
+ new CompositeModificationByteStringPayload(compositeModification.toSerializable());
+ payload.clearModificationReference();
- private CompositeModificationByteStringPayload newByteStringPayload(final Modification... mods) {
- MutableCompositeModification compMod = new MutableCompositeModification();
- for(Modification mod: mods) {
- compMod.addModification(mod);
- }
+ entries.add(new ReplicatedLogImplEntry(0, 1, payload));
- return new CompositeModificationByteStringPayload(compMod.toSerializable());
+ assertNotNull(new AppendEntries(10, "foobar", 10, 10, entries, 10).toSerializable());
}
-
}
package org.opendaylight.controller.cluster.datastore;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
+import org.opendaylight.controller.cluster.raft.RaftVersions;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+@Deprecated
public class CompositeModificationPayloadTest {
-
- private static final String SERIALIZE_OUT = "serialize.out";
-
- @After
- public void shutDown(){
- File f = new File(SERIALIZE_OUT);
- if(f.exists()){
- f.delete();
- }
- }
-
@Test
public void testBasic() throws IOException {
@Override public Payload getData() {
WriteModification writeModification =
new WriteModification(TestModel.TEST_PATH, ImmutableNodes
- .containerNode(TestModel.TEST_QNAME),
- TestModel.createTestContext());
+ .containerNode(TestModel.TEST_QNAME));
MutableCompositeModification compositeModification =
new MutableCompositeModification();
AppendEntries appendEntries =
new AppendEntries(1, "member-1", 0, 100, entries, 1);
- AppendEntriesMessages.AppendEntries o = (AppendEntriesMessages.AppendEntries) appendEntries.toSerializable();
+ AppendEntriesMessages.AppendEntries o = (AppendEntriesMessages.AppendEntries)
+ appendEntries.toSerializable(RaftVersions.HELIUM_VERSION);
- o.writeDelimitedTo(new FileOutputStream(SERIALIZE_OUT));
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ o.writeDelimitedTo(bos);
AppendEntriesMessages.AppendEntries appendEntries2 =
AppendEntriesMessages.AppendEntries
- .parseDelimitedFrom(new FileInputStream(SERIALIZE_OUT));
+ .parseDelimitedFrom(new ByteArrayInputStream(bos.toByteArray()));
AppendEntries appendEntries1 = AppendEntries.fromSerializable(appendEntries2);
Assert.assertTrue(((CompositeModificationPayload) data).getModification().toString().contains(TestModel.TEST_QNAME.getNamespace().toString()));
-
}
-
}
import akka.dispatch.Dispatchers;
import akka.dispatch.OnComplete;
import akka.japi.Creator;
+import akka.japi.Procedure;
import akka.pattern.Patterns;
+import akka.persistence.SnapshotSelectionCriteria;
import akka.testkit.TestActorRef;
import akka.util.Timeout;
import com.google.common.base.Function;
import org.mockito.InOrder;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
+import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
import org.opendaylight.controller.cluster.datastore.utils.InMemoryJournal;
import org.opendaylight.controller.cluster.datastore.utils.InMemorySnapshotStore;
import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
import org.opendaylight.controller.cluster.raft.Snapshot;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
-import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(),
"testApplySnapshot");
- NormalizedNodeToNodeCodec codec =
- new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT);
+ InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
+ store.onGlobalContextUpdated(SCHEMA_CONTEXT);
+
+ writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ YangInstanceIdentifier root = YangInstanceIdentifier.builder().build();
+ NormalizedNode<?,?> expected = readStore(store, root);
+
+ ApplySnapshot applySnapshot = new ApplySnapshot(Snapshot.create(
+ SerializationUtils.serializeNormalizedNode(expected),
+ Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4));
+
+ shard.underlyingActor().onReceiveCommand(applySnapshot);
+
+ NormalizedNode<?,?> actual = readStore(shard, root);
+
+ assertEquals("Root node", expected, actual);
+
+ shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }
+
+ @Test
+ public void testApplyHelium2VersionSnapshot() throws Exception {
+ TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(),
+ "testApplySnapshot");
+
+ NormalizedNodeToNodeCodec codec = new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT);
- writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
+ store.onGlobalContextUpdated(SCHEMA_CONTEXT);
+
+ writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
YangInstanceIdentifier root = YangInstanceIdentifier.builder().build();
- NormalizedNode<?,?> expected = readStore(shard, root);
+ NormalizedNode<?,?> expected = readStore(store, root);
NormalizedNodeMessages.Container encode = codec.encode(expected);
NormalizedNode<?,?> actual = readStore(shard, root);
- assertEquals(expected, actual);
+ assertEquals("Root node", expected, actual);
shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
- MutableCompositeModification compMod = new MutableCompositeModification();
- compMod.addModification(new WriteModification(TestModel.TEST_PATH, node, SCHEMA_CONTEXT));
- Payload payload = new CompositeModificationPayload(compMod.toSerializable());
- ApplyState applyState = new ApplyState(null, "test",
- new ReplicatedLogImplEntry(1, 2, payload));
+ ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
+ newModificationPayload(new WriteModification(TestModel.TEST_PATH, node))));
+
+ shard.underlyingActor().onReceiveCommand(applyState);
+
+ NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH);
+ assertEquals("Applied state", node, actual);
+
+ shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }
+
+ @Test
+ public void testApplyStateLegacy() throws Exception {
+
+ TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyStateLegacy");
+
+ NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
+
+ ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
+ newLegacyByteStringPayload(new WriteModification(TestModel.TEST_PATH, node))));
shard.underlyingActor().onReceiveCommand(applyState);
shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
- @SuppressWarnings("serial")
@Test
public void testRecovery() throws Exception {
InMemoryDOMDataStore testStore = InMemoryDOMDataStoreFactory.create("Test", null, null);
testStore.onGlobalContextUpdated(SCHEMA_CONTEXT);
- DOMStoreWriteTransaction writeTx = testStore.newWriteOnlyTransaction();
- writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- DOMStoreThreePhaseCommitCohort commitCohort = writeTx.ready();
- commitCohort.preCommit().get();
- commitCohort.commit().get();
+ writeToStore(testStore, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- DOMStoreReadTransaction readTx = testStore.newReadOnlyTransaction();
- NormalizedNode<?, ?> root = readTx.read(YangInstanceIdentifier.builder().build()).get().get();
+ NormalizedNode<?, ?> root = readStore(testStore, YangInstanceIdentifier.builder().build());
InMemorySnapshotStore.addSnapshot(shardID.toString(), Snapshot.create(
- new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT).encode(
- root).
+ SerializationUtils.serializeNormalizedNode(root),
+ Collections.<ReplicatedLogEntry>emptyList(), 0, 1, -1, -1));
+
+ // Set up the InMemoryJournal.
+
+ InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newLegacyPayload(
+ new WriteModification(TestModel.OUTER_LIST_PATH,
+ ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()))));
+
+ int nListEntries = 16;
+ Set<Integer> listEntryKeys = new HashSet<>();
+
+ // Add some ModificationPayload entries
+ for(int i = 1; i <= nListEntries; i++) {
+ listEntryKeys.add(Integer.valueOf(i));
+ YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
+ .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
+ Modification mod = new MergeModification(path,
+ ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
+ InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
+ newModificationPayload(mod)));
+ }
+
+ InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
+ new ApplyLogEntries(nListEntries));
+
+ testRecovery(listEntryKeys);
+ }
+
+ @Test
+ public void testHelium2VersionRecovery() throws Exception {
+
+ // Set up the InMemorySnapshotStore.
+
+ InMemoryDOMDataStore testStore = InMemoryDOMDataStoreFactory.create("Test", null, null);
+ testStore.onGlobalContextUpdated(SCHEMA_CONTEXT);
+
+ writeToStore(testStore, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ NormalizedNode<?, ?> root = readStore(testStore, YangInstanceIdentifier.builder().build());
+
+ InMemorySnapshotStore.addSnapshot(shardID.toString(), Snapshot.create(
+ new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT).encode(root).
getNormalizedNode().toByteString().toByteArray(),
Collections.<ReplicatedLogEntry>emptyList(), 0, 1, -1, -1));
// Set up the InMemoryJournal.
- InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newPayload(
+ InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newLegacyPayload(
new WriteModification(TestModel.OUTER_LIST_PATH,
- ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
- SCHEMA_CONTEXT))));
+ ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()))));
int nListEntries = 16;
Set<Integer> listEntryKeys = new HashSet<>();
- for(int i = 1; i <= nListEntries-5; i++) {
+ int i = 1;
+
+ // Add some CompositeModificationPayload entries
+ for(; i <= 8; i++) {
listEntryKeys.add(Integer.valueOf(i));
YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
.nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
Modification mod = new MergeModification(path,
- ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i),
- SCHEMA_CONTEXT);
+ ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
- newPayload(mod)));
+ newLegacyPayload(mod)));
}
- // Add some of the new CompositeModificationByteStringPayload
- for(int i = 11; i <= nListEntries; i++) {
+ // Add some CompositeModificationByteStringPayload entries
+ for(; i <= nListEntries; i++) {
listEntryKeys.add(Integer.valueOf(i));
YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
.nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
Modification mod = new MergeModification(path,
- ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i),
- SCHEMA_CONTEXT);
+ ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
- newByteStringPayload(mod)));
+ newLegacyByteStringPayload(mod)));
}
+ InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1, new ApplyLogEntries(nListEntries));
- InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
- new ApplyLogEntries(nListEntries));
+ testRecovery(listEntryKeys);
+ }
+ private void testRecovery(Set<Integer> listEntryKeys) throws Exception {
// Create the actor and wait for recovery complete.
+ int nListEntries = listEntryKeys.size();
+
final CountDownLatch recoveryComplete = new CountDownLatch(1);
+ @SuppressWarnings("serial")
Creator<Shard> creator = new Creator<Shard>() {
@Override
public Shard create() throws Exception {
shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
- private CompositeModificationPayload newPayload(final Modification... mods) {
+ private CompositeModificationPayload newLegacyPayload(final Modification... mods) {
MutableCompositeModification compMod = new MutableCompositeModification();
for(Modification mod: mods) {
compMod.addModification(mod);
return new CompositeModificationPayload(compMod.toSerializable());
}
- private CompositeModificationByteStringPayload newByteStringPayload(final Modification... mods) {
+ private CompositeModificationByteStringPayload newLegacyByteStringPayload(final Modification... mods) {
MutableCompositeModification compMod = new MutableCompositeModification();
for(Modification mod: mods) {
compMod.addModification(mod);
return new CompositeModificationByteStringPayload(compMod.toSerializable());
}
+ private ModificationPayload newModificationPayload(final Modification... mods) throws IOException {
+ MutableCompositeModification compMod = new MutableCompositeModification();
+ for(Modification mod: mods) {
+ compMod.addModification(mod);
+ }
+
+ return new ModificationPayload(compMod);
+ }
private DOMStoreThreePhaseCommitCohort setupMockWriteTransaction(final String cohortName,
final InMemoryDOMDataStore dataStore, final YangInstanceIdentifier path, final NormalizedNode<?, ?> data,
}
}).when(cohort).abort();
- modification.addModification(new WriteModification(path, data, SCHEMA_CONTEXT));
+ modification.addModification(new WriteModification(path, data));
return cohort;
}
}
@Test
- public void testCreateSnapshot() throws IOException, InterruptedException {
- testCreateSnapshot(true, "testCreateSnapshot");
+ public void testCreateSnapshot() throws Exception {
+ testCreateSnapshot(true, "testCreateSnapshot");
}
@Test
- public void testCreateSnapshotWithNonPersistentData() throws IOException, InterruptedException {
+ public void testCreateSnapshotWithNonPersistentData() throws Exception {
testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData");
}
@SuppressWarnings("serial")
- public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws IOException, InterruptedException {
- final DatastoreContext dataStoreContext = DatastoreContext.newBuilder().
- shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(persistent).build();
+ public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws Exception{
+
+ final AtomicReference<Object> savedSnapshot = new AtomicReference<>();
+ class DelegatingPersistentDataProvider implements DataPersistenceProvider {
+ DataPersistenceProvider delegate;
+
+ DelegatingPersistentDataProvider(DataPersistenceProvider delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public boolean isRecoveryApplicable() {
+ return delegate.isRecoveryApplicable();
+ }
+
+ @Override
+ public <T> void persist(T o, Procedure<T> procedure) {
+ delegate.persist(o, procedure);
+ }
+
+ @Override
+ public void saveSnapshot(Object o) {
+ savedSnapshot.set(o);
+ delegate.saveSnapshot(o);
+ }
+
+ @Override
+ public void deleteSnapshots(SnapshotSelectionCriteria criteria) {
+ delegate.deleteSnapshots(criteria);
+ }
+
+ @Override
+ public void deleteMessages(long sequenceNumber) {
+ delegate.deleteMessages(sequenceNumber);
+ }
+ }
+
+ dataStoreContextBuilder.persistent(persistent);
new ShardTestKit(getSystem()) {{
final AtomicReference<CountDownLatch> latch = new AtomicReference<>(new CountDownLatch(1));
public Shard create() throws Exception {
return new Shard(shardID, Collections.<ShardIdentifier,String>emptyMap(),
newDatastoreContext(), SCHEMA_CONTEXT) {
+
+ DelegatingPersistentDataProvider delegating;
+
+ @Override
+ protected DataPersistenceProvider persistence() {
+ if(delegating == null) {
+ delegating = new DelegatingPersistentDataProvider(super.persistence());
+ }
+
+ return delegating;
+ }
+
@Override
protected void commitSnapshot(final long sequenceNumber) {
super.commitSnapshot(sequenceNumber);
waitUntilLeader(shard);
- shard.tell(new CaptureSnapshot(-1,-1,-1,-1), getRef());
+ writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ NormalizedNode<?,?> expectedRoot = readStore(shard, YangInstanceIdentifier.builder().build());
+
+ CaptureSnapshot capture = new CaptureSnapshot(-1, -1, -1, -1);
+ shard.tell(capture, getRef());
assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
+ assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
+ savedSnapshot.get() instanceof Snapshot);
+
+ verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
+
latch.set(new CountDownLatch(1));
- shard.tell(new CaptureSnapshot(-1,-1,-1,-1), getRef());
+ savedSnapshot.set(null);
+
+ shard.tell(capture, getRef());
assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
+ assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
+ savedSnapshot.get() instanceof Snapshot);
+
+ verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
+
shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ }
+
+ private void verifySnapshot(Snapshot snapshot, NormalizedNode<?,?> expectedRoot) {
+
+ NormalizedNode<?, ?> actual = SerializationUtils.deserializeNormalizedNode(snapshot.getState());
+ assertEquals("Root node", expectedRoot, actual);
+
}};
}
static NormalizedNode<?,?> readStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id)
throws ExecutionException, InterruptedException {
- DOMStoreReadTransaction transaction = shard.underlyingActor().getDataStore().newReadOnlyTransaction();
+ return readStore(shard.underlyingActor().getDataStore(), id);
+ }
+
+ public static NormalizedNode<?,?> readStore(final InMemoryDOMDataStore store, final YangInstanceIdentifier id)
+ throws ExecutionException, InterruptedException {
+ DOMStoreReadTransaction transaction = store.newReadOnlyTransaction();
CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future =
transaction.read(id);
return node;
}
- private void writeToStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id, final NormalizedNode<?,?> node)
- throws ExecutionException, InterruptedException {
- DOMStoreWriteTransaction transaction = shard.underlyingActor().getDataStore().newWriteOnlyTransaction();
+ static void writeToStore(final TestActorRef<Shard> shard, final YangInstanceIdentifier id,
+ final NormalizedNode<?,?> node) throws ExecutionException, InterruptedException {
+ writeToStore(shard.underlyingActor().getDataStore(), id, node);
+ }
+
+ public static void writeToStore(final InMemoryDOMDataStore store, final YangInstanceIdentifier id,
+ final NormalizedNode<?,?> node) throws ExecutionException, InterruptedException {
+ DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
transaction.write(id, node);
import com.google.common.util.concurrent.MoreExecutors;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
-import org.junit.BeforeClass;
+import org.junit.Before;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.ShardWriteTransaction.GetCompositeModificationReply;
import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateSnapshot;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec.Encoded;
+import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
+import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import scala.concurrent.duration.Duration;
public class ShardTransactionTest extends AbstractActorTest {
- private static final InMemoryDOMDataStore store =
- new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
private static final SchemaContext testSchemaContext = TestModel.createTestContext();
private final ShardStats shardStats = new ShardStats(SHARD_IDENTIFIER.toString(), "DataStore");
- @BeforeClass
- public static void staticSetup() {
+ private final InMemoryDOMDataStore store =
+ new InMemoryDOMDataStore("OPER", MoreExecutors.sameThreadExecutor());
+
+ @Before
+ public void setup() {
store.onGlobalContextUpdated(testSchemaContext);
}
Collections.<ShardIdentifier, String>emptyMap(), datastoreContext, TestModel.createTestContext()));
}
+ private ActorRef newTransactionActor(DOMStoreTransaction transaction, String name) {
+ return newTransactionActor(transaction, name, DataStoreVersions.CURRENT_VERSION);
+ }
+
+ private ActorRef newTransactionActor(DOMStoreTransaction transaction, String name, short version) {
+ return newTransactionActor(transaction, null, name, version);
+ }
+
+ private ActorRef newTransactionActor(DOMStoreTransaction transaction, ActorRef shard, String name) {
+ return newTransactionActor(transaction, null, name, DataStoreVersions.CURRENT_VERSION);
+ }
+
+ private ActorRef newTransactionActor(DOMStoreTransaction transaction, ActorRef shard, String name,
+ short version) {
+ Props props = ShardTransaction.props(transaction, shard != null ? shard : createShard(),
+ testSchemaContext, datastoreContext, shardStats, "txn", version);
+ return getSystem().actorOf(props, name);
+ }
+
@Test
public void testOnReceiveReadData() throws Exception {
new JavaTestKit(getSystem()) {{
final ActorRef shard = createShard();
- Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- testOnReceiveReadData(getSystem().actorOf(props, "testReadDataRO"));
+ testOnReceiveReadData(newTransactionActor(store.newReadOnlyTransaction(), shard, "testReadDataRO"));
- props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
-
- testOnReceiveReadData(getSystem().actorOf(props, "testReadDataRW"));
+ testOnReceiveReadData(newTransactionActor(store.newReadWriteTransaction(), shard, "testReadDataRW"));
}
private void testOnReceiveReadData(final ActorRef transaction) {
public void testOnReceiveReadDataWhenDataNotFound() throws Exception {
new JavaTestKit(getSystem()) {{
final ActorRef shard = createShard();
- Props props = ShardTransaction.props( store.newReadOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
-
- testOnReceiveReadDataWhenDataNotFound(getSystem().actorOf(
- props, "testReadDataWhenDataNotFoundRO"));
- props = ShardTransaction.props( store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
+ testOnReceiveReadDataWhenDataNotFound(newTransactionActor(
+ store.newReadOnlyTransaction(), shard, "testReadDataWhenDataNotFoundRO"));
- testOnReceiveReadDataWhenDataNotFound(getSystem().actorOf(
- props, "testReadDataWhenDataNotFoundRW"));
+ testOnReceiveReadDataWhenDataNotFound(newTransactionActor(
+ store.newReadWriteTransaction(), shard, "testReadDataWhenDataNotFoundRW"));
}
private void testOnReceiveReadDataWhenDataNotFound(final ActorRef transaction) {
@Test
public void testOnReceiveReadDataHeliumR1() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.HELIUM_1_VERSION);
-
- ActorRef transaction = getSystem().actorOf(props, "testOnReceiveReadDataHeliumR1");
+ ActorRef transaction = newTransactionActor(store.newReadOnlyTransaction(),
+ "testOnReceiveReadDataHeliumR1", DataStoreVersions.HELIUM_1_VERSION);
transaction.tell(new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(),
getRef());
public void testOnReceiveDataExistsPositive() throws Exception {
new JavaTestKit(getSystem()) {{
final ActorRef shard = createShard();
- Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
-
- testOnReceiveDataExistsPositive(getSystem().actorOf(props, "testDataExistsPositiveRO"));
- props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
+ testOnReceiveDataExistsPositive(newTransactionActor(store.newReadOnlyTransaction(), shard,
+ "testDataExistsPositiveRO"));
- testOnReceiveDataExistsPositive(getSystem().actorOf(props, "testDataExistsPositiveRW"));
+ testOnReceiveDataExistsPositive(newTransactionActor(store.newReadWriteTransaction(), shard,
+ "testDataExistsPositiveRW"));
}
private void testOnReceiveDataExistsPositive(final ActorRef transaction) {
public void testOnReceiveDataExistsNegative() throws Exception {
new JavaTestKit(getSystem()) {{
final ActorRef shard = createShard();
- Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- testOnReceiveDataExistsNegative(getSystem().actorOf(props, "testDataExistsNegativeRO"));
+ testOnReceiveDataExistsNegative(newTransactionActor(store.newReadOnlyTransaction(), shard,
+ "testDataExistsNegativeRO"));
- props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
-
- testOnReceiveDataExistsNegative(getSystem().actorOf(props, "testDataExistsNegativeRW"));
+ testOnReceiveDataExistsNegative(newTransactionActor(store.newReadWriteTransaction(), shard,
+ "testDataExistsNegativeRW"));
}
private void testOnReceiveDataExistsNegative(final ActorRef transaction) {
@Test
public void testOnReceiveWriteData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newWriteOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testOnReceiveWriteData");
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveWriteData");
transaction.tell(new WriteData(TestModel.TEST_PATH,
ImmutableNodes.containerNode(TestModel.TEST_QNAME)).toSerializable(
@Test
public void testOnReceiveHeliumR1WriteData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newWriteOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.HELIUM_1_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testOnReceiveHeliumR1WriteData");
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveHeliumR1WriteData", DataStoreVersions.HELIUM_1_VERSION);
Encoded encoded = new NormalizedNodeToNodeCodec(null).encode(TestModel.TEST_PATH,
ImmutableNodes.containerNode(TestModel.TEST_QNAME));
@Test
public void testOnReceiveMergeData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testMergeData");
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testMergeData");
transaction.tell(new MergeData(TestModel.TEST_PATH,
ImmutableNodes.containerNode(TestModel.TEST_QNAME)).toSerializable(
@Test
public void testOnReceiveHeliumR1MergeData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newWriteOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.HELIUM_1_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testOnReceiveHeliumR1MergeData");
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveHeliumR1MergeData", DataStoreVersions.HELIUM_1_VERSION);
Encoded encoded = new NormalizedNodeToNodeCodec(null).encode(TestModel.TEST_PATH,
ImmutableNodes.containerNode(TestModel.TEST_QNAME));
@Test
public void testOnReceiveDeleteData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props( store.newWriteOnlyTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testDeleteData");
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testDeleteData");
transaction.tell(new DeleteData(TestModel.TEST_PATH).toSerializable(
DataStoreVersions.HELIUM_2_VERSION), getRef());
@Test
public void testOnReceiveReadyTransaction() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props( store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testReadyTransaction");
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testReadyTransaction");
watch(transaction);
// test
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props( store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testReadyTransaction2");
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testReadyTransaction2");
watch(transaction);
expectMsgAnyClassOf(duration("5 seconds"), ReadyTransactionReply.class,
Terminated.class);
}};
+ }
+ @Test
+ public void testOnReceiveCreateSnapshot() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ ShardTest.writeToStore(store, TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ NormalizedNode<?,?> expectedRoot = ShardTest.readStore(store,
+ YangInstanceIdentifier.builder().build());
+
+ final ActorRef transaction = newTransactionActor(store.newReadOnlyTransaction(),
+ "testOnReceiveCreateSnapshot");
+
+ watch(transaction);
+
+ transaction.tell(CreateSnapshot.INSTANCE, getRef());
+
+ CaptureSnapshotReply reply = expectMsgClass(duration("3 seconds"), CaptureSnapshotReply.class);
+
+ assertNotNull("getSnapshot is null", reply.getSnapshot());
+
+ NormalizedNode<?,?> actualRoot = SerializationUtils.deserializeNormalizedNode(
+ reply.getSnapshot());
+
+ assertEquals("Root node", expectedRoot, actualRoot);
+
+ expectTerminated(duration("3 seconds"), transaction);
+ }};
}
- @SuppressWarnings("unchecked")
@Test
public void testOnReceiveCloseTransaction() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction = getSystem().actorOf(props, "testCloseTransaction");
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testCloseTransaction");
watch(transaction);
Duration.create(500, TimeUnit.MILLISECONDS)).build();
new JavaTestKit(getSystem()) {{
- final ActorRef shard = createShard();
- final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
- testSchemaContext, datastoreContext, shardStats, "txn",
- DataStoreVersions.CURRENT_VERSION);
- final ActorRef transaction =
- getSystem().actorOf(props, "testShardTransactionInactivity");
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testShardTransactionInactivity");
watch(transaction);
package org.opendaylight.controller.cluster.datastore.modification;
+import static org.junit.Assert.assertEquals;
import com.google.common.base.Optional;
+import org.apache.commons.lang.SerializationUtils;
import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
-public class DeleteModificationTest extends AbstractModificationTest{
-
- @Test
- public void testApply() throws Exception {
- //Write something into the datastore
- DOMStoreReadWriteTransaction writeTransaction = store.newReadWriteTransaction();
- WriteModification writeModification = new WriteModification(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext());
- writeModification.apply(writeTransaction);
- commitTransaction(writeTransaction);
-
- //Check if it's in the datastore
- Optional<NormalizedNode<?,?>> data = readData(TestModel.TEST_PATH);
- Assert.assertTrue(data.isPresent());
-
- //Delete stuff from the datastore
- DOMStoreWriteTransaction deleteTransaction = store.newWriteOnlyTransaction();
- DeleteModification deleteModification = new DeleteModification(TestModel.TEST_PATH);
- deleteModification.apply(deleteTransaction);
- commitTransaction(deleteTransaction);
-
- data = readData(TestModel.TEST_PATH);
- Assert.assertFalse(data.isPresent());
- }
+public class DeleteModificationTest extends AbstractModificationTest {
+
+ @Test
+ public void testApply() throws Exception {
+ // Write something into the datastore
+ DOMStoreReadWriteTransaction writeTransaction = store.newReadWriteTransaction();
+ WriteModification writeModification = new WriteModification(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ writeModification.apply(writeTransaction);
+ commitTransaction(writeTransaction);
+
+ // Check if it's in the datastore
+ Optional<NormalizedNode<?, ?>> data = readData(TestModel.TEST_PATH);
+ Assert.assertTrue(data.isPresent());
+
+ // Delete stuff from the datastore
+ DOMStoreWriteTransaction deleteTransaction = store.newWriteOnlyTransaction();
+ DeleteModification deleteModification = new DeleteModification(TestModel.TEST_PATH);
+ deleteModification.apply(deleteTransaction);
+ commitTransaction(deleteTransaction);
+
+ data = readData(TestModel.TEST_PATH);
+ Assert.assertFalse(data.isPresent());
+ }
+
+ @Test
+ public void testSerialization() {
+ YangInstanceIdentifier path = TestModel.TEST_PATH;
+
+ DeleteModification expected = new DeleteModification(path);
+
+ DeleteModification clone = (DeleteModification) SerializationUtils.clone(expected);
+ assertEquals("getPath", expected.getPath(), clone.getPath());
+ }
}
package org.opendaylight.controller.cluster.datastore.modification;
+import static org.junit.Assert.assertEquals;
import com.google.common.base.Optional;
+import org.apache.commons.lang.SerializationUtils;
import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
public class MergeModificationTest extends AbstractModificationTest{
//Write something into the datastore
DOMStoreReadWriteTransaction writeTransaction = store.newReadWriteTransaction();
- MergeModification writeModification = new MergeModification(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext());
+ MergeModification writeModification = new MergeModification(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
writeModification.apply(writeTransaction);
commitTransaction(writeTransaction);
@Test
public void testSerialization() {
- SchemaContext schemaContext = TestModel.createTestContext();
- NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
- MergeModification mergeModification = new MergeModification(TestModel.TEST_PATH,
- node, schemaContext);
+ YangInstanceIdentifier path = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> data = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.DESC_QNAME, "foo")).build();
- Object serialized = mergeModification.toSerializable();
+ MergeModification expected = new MergeModification(path, data);
- MergeModification newModification = MergeModification.fromSerializable(serialized, schemaContext);
-
- Assert.assertEquals("getPath", TestModel.TEST_PATH, newModification.getPath());
- Assert.assertEquals("getData", node, newModification.getData());
+ MergeModification clone = (MergeModification) SerializationUtils.clone(expected);
+ assertEquals("getPath", expected.getPath(), clone.getPath());
+ assertEquals("getData", expected.getData(), clone.getData());
}
}
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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.datastore.modification;
+
+import static org.junit.Assert.assertEquals;
+import org.apache.commons.lang.SerializationUtils;
+import org.junit.Test;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
+
+/**
+ * Unit tests for ModificationPayload.
+ *
+ * @author Thomas Pantelis
+ */
+public class ModificationPayloadTest {
+
+ @Test
+ public void test() throws Exception {
+
+ YangInstanceIdentifier writePath = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> writeData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.DESC_QNAME, "foo")).build();
+
+ MutableCompositeModification compositeModification = new MutableCompositeModification();
+ compositeModification.addModification(new WriteModification(writePath, writeData));
+
+ ModificationPayload payload = new ModificationPayload(compositeModification);
+
+ MutableCompositeModification deserialized = (MutableCompositeModification) payload.getModification();
+
+ assertEquals("getModifications size", 1, deserialized.getModifications().size());
+ WriteModification write = (WriteModification)deserialized.getModifications().get(0);
+ assertEquals("getPath", writePath, write.getPath());
+ assertEquals("getData", writeData, write.getData());
+
+ ModificationPayload cloned =
+ (ModificationPayload) SerializationUtils.clone(payload);
+
+ deserialized = (MutableCompositeModification) payload.getModification();
+
+ assertEquals("getModifications size", 1, deserialized.getModifications().size());
+ write = (WriteModification)deserialized.getModifications().get(0);
+ assertEquals("getPath", writePath, write.getPath());
+ assertEquals("getData", writeData, write.getData());
+ }
+}
package org.opendaylight.controller.cluster.datastore.modification;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import com.google.common.base.Optional;
+import com.google.common.base.Stopwatch;
+import org.apache.commons.lang.SerializationUtils;
+import org.junit.Ignore;
import org.junit.Test;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
public class MutableCompositeModificationTest extends AbstractModificationTest {
MutableCompositeModification compositeModification = new MutableCompositeModification();
compositeModification.addModification(new WriteModification(TestModel.TEST_PATH,
- ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext()));
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)));
DOMStoreReadWriteTransaction transaction = store.newReadWriteTransaction();
compositeModification.apply(transaction);
}
@Test
- public void testEverySerializedCompositeModificationObjectMustBeDifferent(){
+ public void testSerialization() {
+ YangInstanceIdentifier writePath = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> writeData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.DESC_QNAME, "foo")).build();
+
+ YangInstanceIdentifier mergePath = TestModel.OUTER_LIST_PATH;
+ NormalizedNode<?, ?> mergeData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.OUTER_LIST_QNAME)).build();
+
+ YangInstanceIdentifier deletePath = TestModel.TEST_PATH;
+
MutableCompositeModification compositeModification = new MutableCompositeModification();
- compositeModification.addModification(new WriteModification(TestModel.TEST_PATH,
- ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext()));
- Object one = compositeModification.toSerializable();
- try{Thread.sleep(10);}catch(Exception err){}
- Object two = compositeModification.toSerializable();
- assertNotEquals(one,two);
+ compositeModification.addModification(new WriteModification(writePath, writeData));
+ compositeModification.addModification(new MergeModification(mergePath, mergeData));
+ compositeModification.addModification(new DeleteModification(deletePath));
+
+ MutableCompositeModification clone = (MutableCompositeModification) SerializationUtils.clone(compositeModification);
+
+ assertEquals("getModifications size", 3, clone.getModifications().size());
+
+ WriteModification write = (WriteModification)clone.getModifications().get(0);
+ assertEquals("getPath", writePath, write.getPath());
+ assertEquals("getData", writeData, write.getData());
+
+ MergeModification merge = (MergeModification)clone.getModifications().get(1);
+ assertEquals("getPath", mergePath, merge.getPath());
+ assertEquals("getData", mergeData, merge.getData());
+
+ DeleteModification delete = (DeleteModification)clone.getModifications().get(2);
+ assertEquals("getPath", deletePath, delete.getPath());
+ }
+
+ @Test
+ @Ignore
+ public void testSerializationScale() throws Exception {
+ YangInstanceIdentifier writePath = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> writeData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.DESC_QNAME, "foo")).build();
+
+ MutableCompositeModification compositeModification = new MutableCompositeModification();
+ for(int i = 0; i < 1000; i++) {
+ compositeModification.addModification(new WriteModification(writePath, writeData));
+ }
+
+ Stopwatch sw = new Stopwatch();
+ sw.start();
+ for(int i = 0; i < 1000; i++) {
+ new ModificationPayload(compositeModification);
+ }
+
+ sw.stop();
+ System.out.println("Elapsed: "+sw);
+
+ ModificationPayload p = new ModificationPayload(compositeModification);
+ sw.start();
+ for(int i = 0; i < 1000; i++) {
+ p.getModification();
+ }
+
+ sw.stop();
+ System.out.println("Elapsed: "+sw);
}
}
package org.opendaylight.controller.cluster.datastore.modification;
+import static org.junit.Assert.assertEquals;
import com.google.common.base.Optional;
+import org.apache.commons.lang.SerializationUtils;
import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
-import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
-public class WriteModificationTest extends AbstractModificationTest{
+public class WriteModificationTest extends AbstractModificationTest {
@Test
public void testApply() throws Exception {
//Write something into the datastore
DOMStoreReadWriteTransaction writeTransaction = store.newReadWriteTransaction();
WriteModification writeModification = new WriteModification(TestModel.TEST_PATH,
- ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext());
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
writeModification.apply(writeTransaction);
commitTransaction(writeTransaction);
@Test
public void testSerialization() {
- SchemaContext schemaContext = TestModel.createTestContext();
- NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
- WriteModification writeModification = new WriteModification(TestModel.TEST_PATH,
- node, schemaContext);
-
- Object serialized = writeModification.toSerializable();
+ YangInstanceIdentifier path = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> data = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(TestModel.TEST_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.DESC_QNAME, "foo")).build();
- WriteModification newModification = WriteModification.fromSerializable(serialized, schemaContext);
+ WriteModification expected = new WriteModification(path, data);
- Assert.assertEquals("getPath", TestModel.TEST_PATH, newModification.getPath());
- Assert.assertEquals("getData", node, newModification.getData());
+ WriteModification clone = (WriteModification) SerializationUtils.clone(expected);
+ assertEquals("getPath", expected.getPath(), clone.getPath());
+ assertEquals("getData", expected.getData(), clone.getData());
}
}
import org.opendaylight.controller.cluster.example.messages.KeyValue;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
@Override public Payload getData() {
WriteModification writeModification =
new WriteModification(TestModel.TEST_PATH, ImmutableNodes
- .containerNode(TestModel.TEST_QNAME),
- TestModel.createTestContext()
- );
+ .containerNode(TestModel.TEST_QNAME));
MutableCompositeModification compositeModification =
new MutableCompositeModification();
compositeModification.addModification(writeModification);
- return new CompositeModificationPayload(
+ return new CompositeModificationByteStringPayload(
compositeModification.toSerializable());
}
public static class ServerActor extends UntypedActor {
@Override public void onReceive(Object message) throws Exception {
- if(AppendEntries.SERIALIZABLE_CLASS.equals(message.getClass())){
+ if(AppendEntries.LEGACY_SERIALIZABLE_CLASS.equals(message.getClass())){
AppendEntries appendEntries =
AppendEntries.fromSerializable(message);
@Override
public void onCarBought(CarBought notification) {
- log.info("onCarBought notification : Adding car person entry");
final CarPersonBuilder carPersonBuilder = new CarPersonBuilder();
carPersonBuilder.setCarId(notification.getCarId());
carPersonBuilder.setKey(key);
final CarPerson carPerson = carPersonBuilder.build();
+ log.info("Car bought, adding car-person entry: [{}]", carPerson);
+
InstanceIdentifier<CarPerson> carPersonIId =
InstanceIdentifier.<CarPeople>builder(CarPeople.class).child(CarPerson.class, carPerson.getKey()).build();
Futures.addCallback(tx.submit(), new FutureCallback<Void>() {
@Override
public void onSuccess(final Void result) {
- log.info("Car bought, entry added to map of people and car [{}]", carPerson);
+ log.info("Successfully added car-person entry: [{}]", carPerson);
}
@Override
public void onFailure(final Throwable t) {
- log.info("Car bought, Failed entry addition to map of people and car [{}]", carPerson);
+ log.error(String.format("Failed to add car-person entry: [%s]", carPerson), t);
}
});
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
+import java.util.concurrent.Future;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.people.rev140818.PersonContext;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.people.rev140818.people.Person;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.config.sal.clustering.it.people.rev140818.people.PersonBuilder;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.common.RpcError;
import org.opendaylight.yangtools.yang.common.RpcResult;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.concurrent.Future;
-
public class PeopleProvider implements PeopleService, AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(PeopleProvider.class);
@Override
public void onFailure(final Throwable t) {
- log.info("RPC addPerson : person addition failed [{}]", person);
+ log.error(String.format("RPC addPerson : person addition failed [%s]", person), t);
futureResult.set(RpcResultBuilder.<Void>failed()
.withError(RpcError.ErrorType.APPLICATION, t.getMessage()).build());
}
Code Review / controller.git / commitdiff