Creating an Application based on MD-SAL(Data Store, RPC, Notification )

Data Store 

• Yang data is a tree- all staterelated data are modeled and represented as data tree to address any element / subtree
• Two Logical Data Stores
   • config 
   • operational
• Unified View
• InstanceIdentifier:
   • Pointer to a node

Data Store Transactions 

• Multicast asynchronous communications, sent by Data Broker if there is change in conceptual data tree, and is delivered to subscribed consumers

• Data Stores:
   • Operational Data Tree - published by the providers using MD-SAL, represents a feedback loop for applications to observe state of the network / system.
   • Configuration Data Tree - populated by consumers, represents intended state of the system / network预期状态

• Transaction Types
   • Transactional modification to conceptual data tree - write transactions         newWriteOnlyTransaction()
   • Transactional reads from current data tree and not affected by any      
      subsequent write - read-only transactions newReadOnlyTransaction()
   • Transaction provides both read and write capabilities – read-write  
      transactions newReadWriteTransaction()

Modifications on Data tree:
   • Put – stores a piece of data on specified path, act as add/replace
   • void    put(LogicalDatastoreType store, InstanceIdentifier<T> path, T            data);
   • Merge – merges a piece of data on the existing data on specified path
   • void merge(LogicalDatastoreType store, InstanceIdentifier<T> path, T
     data);
   • Delete – removes the whole subtree from the specified path
   • void delete(LogicalDatastoreType store, InstanceIdentifier<?> path);





Simple Data Transaction Steps 

Let assume initial state of data tree for PATH is A

1. Allocates new ReadWriteTransaction
   ReadWriteTransaction rwTx = broker.newReadWriteTransaction();

2. Read from rwTx will return value A for PATH
   rwRx.read(OPERATIONAL,PATH).get();

3. Writes value B to PATH using rwTx
   rwRx.put(OPERATIONAL,PATH,B);

4. Read will return value B for PATH, since previous write occurred in same transaction
   rwRx.read(OPERATIONAL,PATH).get();

5. Writes value C to PATH using rwTx
   rwRx.put(OPERATIONAL,PATH,C);

6. Read will return value C for PATH, since previous write occurred in same transaction
   rwRx.read(OPERATIONAL,PATH).get();



Data Store – Transactions – Reading and Writing

ReadWriteTransaction transaction = dataBroker.newReadWriteTransaction();
Optional<Node> nodeOptional;
nodeOptional = transaction.read( LogicalDataStore.OPERATIONAL, n1InstanceIdentifier);
transaction.put( LogicalDataStore.CONFIG, n2InstanceIdentifier, topologyNodeBuilder.build());
transaction.delete( LogicalDataStore.CONFIG, n3InstanceIdentifier);
CheckedFuture future;
future = transaction.submit();






E.g: HelloWorld & OpenFlow applications

 public HelloWorldImpl(DataBroker db) {
    this.db = db;
    initializeDataTree(this.db);
}
WriteOnlyTransaction transaction = dataBroker.newWriteOnlyTransaction();
InstanceIdentifier<Node> path = 
  InstanceIdentifier 
  .create(NetworkTopology.class) 
  .child(Topology.class, 
          new TopologyKey(“overlay1”));
transaction.put( 
  LogicalDataStore.CONFIG, 
  path, 
  topologyBuilder.build());
CheckedFuture future; 
future = transaction.submit();


openflowplugin/openflowpluginimpl/src/main/java/org/opendaylight/openflowplugin/impl/OpenFlowPluginProvi derImpl.java 

@Override 
public void setDataBroker(final DataBroker dataBroker) 
{ 
   this.dataBroker = dataBroker;
}
ReadWriteTransaction modification = dataBroker.newReadWriteTransaction(); InstanceIdentifier<Group> path1 =
   InstanceIdentifier
   .create(Nodes.class) 
   .child(Node.class, testNode12.getKey())
   .augmentation(FlowCapableNode.class) 
   .child(Group.class, new GroupKey(group.getGroupId())); 

modification.merge(LogicalDatastoreType.CONFIGURATION,   
                                nodeToInstanceId(testNode12), 
                                testNode12, true); modification.merge(LogicalDatastoreType.CONFIGURATION, 
                                 path1, group, true); 
CheckedFuture<Void, TransactionCommitFailedException>
commitFuture = modification.submit();











Notifications

• Represent asynchronous events with multicast communication, published by providers for consumers
• Used to model events originating in network devices or southbound plugins that are exposed to consumers for listening
• Defined with the notification statement:

  notification
 {
   ….
 }
• NotificationService – is a notification broker that allows clients to subscribe and publish YANG-modeled notifications
• NotificationPublishService.java - interface to publish any YANG-modeled notification which will be delivered to all subscribed listeners




E.g. OpenFlow application



import org.opendaylight.controller.md.sal.binding.api .NotificationPublishService;
import org.opendaylight.controller.md.sal.binding.api .NotificationService;

@Override
public void setNotificationProviderService(final NotificationService notificationProviderService)
{
  this.notificationProviderService =notificationProviderService;
}
@Override
public void setNotificationPublishService(final NotificationPublishService notificationPublishProviderService)
{
   this.notificationPublishService =notificationPublishProviderService;
}



Openflowplugin/testprovider/src/main/java/org/opendaylight/openflowplugin/test/OpenflowpluginTestCommandProvider.java


public void onSessionInitiated(final ProviderContext session) {
  notificationService =session.getSALService(NotificationService.class);
// For switch events
notificationService.registerNotificationListener(flowEventListener); notificationService.registerNotificationListener(nodeErrorListener);
dataBroker = session.getSALService(DataBroker.class);
ctx.registerService(CommandProvider.class.getName(), this, null); createTestFlow(createTestNode(null),null, null);
}




E.g. OpenFlow Pipeline Processing

Remote Procedure Calls (RPCs)

 Used for any call or invocation that crosses the plugin or module boundaries
• Triggered by consumers (Restconf, Rest)
• Communication is Unicast between consumer and provider
• Consumer sends request message to provider responding with reply message
• Models any procedure call implemented by a Provider (Server) exposing functionality to Consumers (Clients)
• Two types of RPCs –
• Global – one service instance per controller container
• Routed – multiple service instance per controller container




RPCs –Sending a Message
E.g:HelloWorld application



HelloServicehelloService=
session.getRpcService(HelloService.class);
Future<RpcResult<HelloWorldOutput>> future;
future= helloService
             .helloWorld(helloWorldInput);
HelloWorldOutput helloWorldOutput= future.get().getResult();

Global RPCs –processing a message –Sync HelloWorld application



public class HelloWorldImpl implements HelloService{
public HelloWorldImpl(ProviderContextsession){
     session.addRpcImplementation(HelloService.class,this);
}
@Override
public Future<RpcResult<HelloWorldOutput>> helloWorld(HelloWorldInputinput) {
/* construct output */
return RpcResultBuilder.success(helloWorldOutput).buildFuture();
}
}

Routed RPCs –E.g:HelloWorld application



public class HelloWorldImpl1 implements HelloService{
public HelloWorldImpl(ProviderContextsession){
   RoutedRpcRegistration<HelloService> reg1=session.addRoutedRpcImplementation  (HelloService.class, this);
  reg1.registerPath(MyContext.class,iid1);
}
/* helloWorld() implementation works as before */
}
public class HelloWorldImpl2 implements HelloService{
public HelloWorldImpl(ProviderContextsession){
  RoutedRpcRegistration<HelloService> reg2 =
  session.addRoutedRpcImplementation(HelloService.class, this);
  reg2.registerPath(MyContext.class,iid2);
}
/* helloWorld() implementation works as before */
}







 3 Brokers

Take-away: for Application Development 

• RPCs:
   registerToHandle( RPC, implementation )
   call( RPC, input ) => output

• Data Store:
   put( key, value )
   get( key ) => value
 
• Notifications:
   subscribe( notification, callback )
   create( notification, data )