+ private StaleFlow makeStaleFlow(InstanceIdentifier<Flow> identifier, Flow del,
+ InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ StaleFlowBuilder staleFlowBuilder = new StaleFlowBuilder(del);
+ return staleFlowBuilder.setId(del.getId()).build();
+ }
+
+ private void persistStaleFlow(StaleFlow staleFlow, InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
+ writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getStaleFlowInstanceIdentifier(staleFlow, nodeIdent),
+ staleFlow, false);
+
+ FluentFuture<?> submitFuture = writeTransaction.commit();
+ handleStaleFlowResultFuture(submitFuture);
+ }
+
+ private void handleStaleFlowResultFuture(FluentFuture<?> submitFuture) {
+ submitFuture.addCallback(new FutureCallback<Object>() {
+ @Override
+ public void onSuccess(Object result) {
+ LOG.debug("Stale Flow creation success");
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ LOG.error("Stale Flow creation failed", throwable);
+ }
+ }, MoreExecutors.directExecutor());
+
+ }
+
+ private InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight
+ .flow.inventory.rev130819.tables.table.StaleFlow> getStaleFlowInstanceIdentifier(
+ StaleFlow staleFlow, InstanceIdentifier<FlowCapableNode> nodeIdent) {
+ return nodeIdent.child(Table.class, new TableKey(staleFlow.getTableId())).child(
+ org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.StaleFlow.class,
+ new StaleFlowKey(new FlowId(staleFlow.getId())));
+ }
+
+ private ListenableFuture<RpcResult<AddGroupOutput>> pushDependentGroup(
+ final InstanceIdentifier<FlowCapableNode> nodeIdent, final Uint32 groupId) {
+
+ //TODO This read to the DS might have a performance impact.
+ //if the dependent group is not installed than we should just cache the parent group,
+ //till we receive the dependent group DTCN and then push it.
+
+ InstanceIdentifier<Group> groupIdent = buildGroupInstanceIdentifier(nodeIdent, groupId);
+ ListenableFuture<RpcResult<AddGroupOutput>> resultFuture;
+ LOG.info("Reading the group from config inventory: {}", groupId);
+ try (ReadTransaction readTransaction = provider.getReadTransaction()) {
+ Optional<Group> group = readTransaction.read(LogicalDatastoreType.CONFIGURATION, groupIdent).get();
+ if (group.isPresent()) {
+ final AddGroupInputBuilder builder = new AddGroupInputBuilder(group.get());
+ builder.setNode(new NodeRef(nodeIdent.firstIdentifierOf(Node.class)));
+ builder.setGroupRef(new GroupRef(nodeIdent));
+ builder.setTransactionUri(new Uri(provider.getNewTransactionId()));
+ AddGroupInput addGroupInput = builder.build();
+ resultFuture = this.provider.getSalGroupService().addGroup(addGroupInput);
+ } else {
+ resultFuture = Futures.immediateFuture(RpcResultBuilder.<AddGroupOutput>failed()
+ .withError(RpcError.ErrorType.APPLICATION,
+ "Group " + groupId + " not present in the config inventory").build());
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.error("Error while reading group from config datastore for the group ID {}", groupId, e);
+ resultFuture = Futures.immediateFuture(RpcResultBuilder.<AddGroupOutput>failed()
+ .withError(RpcError.ErrorType.APPLICATION,
+ "Error while reading group " + groupId + " from inventory").build());
+ }
+ return resultFuture;
+ }
+
+ private final class AddFlowCallBack implements FutureCallback<RpcResult<AddGroupOutput>> {
+ private final AddFlowInput addFlowInput;
+ private final NodeId nodeId;
+ private final Uint32 groupId;
+ private final SettableFuture<RpcResult<AddFlowOutput>> resultFuture;
+
+ private AddFlowCallBack(final AddFlowInput addFlowInput, final NodeId nodeId, Uint32 groupId,
+ SettableFuture<RpcResult<AddFlowOutput>> resultFuture) {
+ this.addFlowInput = addFlowInput;
+ this.nodeId = nodeId;
+ this.groupId = groupId;
+ this.resultFuture = resultFuture;
+ }
+
+ @Override
+ public void onSuccess(RpcResult<AddGroupOutput> rpcResult) {
+ if (rpcResult.isSuccessful() || rpcResult.getErrors().size() == 1
+ && rpcResult.getErrors().iterator().next().getMessage().contains(GROUP_EXISTS_IN_DEVICE_ERROR)) {
+ provider.getDevicesGroupRegistry().storeGroup(nodeId, groupId);
+ Futures.addCallback(provider.getSalFlowService().addFlow(addFlowInput),
+ new FutureCallback<RpcResult<AddFlowOutput>>() {
+ @Override
+ public void onSuccess(RpcResult<AddFlowOutput> result) {
+ resultFuture.set(result);
+ }
+
+ @Override
+ public void onFailure(Throwable failure) {
+ resultFuture.setException(failure);
+ }
+ }, MoreExecutors.directExecutor());
+
+ LOG.debug("Flow add with id {} finished without error for node {}",
+ getFlowId(addFlowInput.getFlowRef()), nodeId);
+ } else {
+ LOG.error("Flow add with id {} failed for node {} with error {}", getFlowId(addFlowInput.getFlowRef()),
+ nodeId, rpcResult.getErrors());
+ resultFuture.set(RpcResultBuilder.<AddFlowOutput>failed()
+ .withRpcErrors(rpcResult.getErrors()).build());
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ LOG.error("Service call for adding flow with id {} failed for node {}",
+ getFlowId(addFlowInput.getFlowRef()), nodeId, throwable);
+ resultFuture.setException(throwable);
+ }
+ }
+
+ private final class UpdateFlowCallBack implements FutureCallback<RpcResult<AddGroupOutput>> {
+ private final UpdateFlowInput updateFlowInput;
+ private final NodeId nodeId;
+ private final Uint32 groupId;
+ private final SettableFuture<RpcResult<UpdateFlowOutput>> resultFuture;
+
+ private UpdateFlowCallBack(final UpdateFlowInput updateFlowInput, final NodeId nodeId,
+ SettableFuture<RpcResult<UpdateFlowOutput>> resultFuture, Uint32 groupId) {
+ this.updateFlowInput = updateFlowInput;
+ this.nodeId = nodeId;
+ this.groupId = groupId;
+ this.resultFuture = resultFuture;
+ }
+
+ @Override
+ public void onSuccess(RpcResult<AddGroupOutput> rpcResult) {
+ if (rpcResult.isSuccessful() || rpcResult.getErrors().size() == 1
+ && rpcResult.getErrors().iterator().next().getMessage().contains(GROUP_EXISTS_IN_DEVICE_ERROR)) {
+ provider.getDevicesGroupRegistry().storeGroup(nodeId, groupId);
+ Futures.addCallback(provider.getSalFlowService().updateFlow(updateFlowInput),
+ new FutureCallback<RpcResult<UpdateFlowOutput>>() {
+ @Override
+ public void onSuccess(RpcResult<UpdateFlowOutput> result) {
+ resultFuture.set(result);
+ }
+
+ @Override
+ public void onFailure(Throwable failure) {
+ resultFuture.setException(failure);
+ }
+ }, MoreExecutors.directExecutor());
+
+ LOG.debug("Flow update with id {} finished without error for node {}",
+ getFlowId(updateFlowInput.getFlowRef()), nodeId);
+ } else {
+ LOG.error("Flow update with id {} failed for node {} with error {}",
+ getFlowId(updateFlowInput.getFlowRef()), nodeId, rpcResult.getErrors());
+ resultFuture.set(RpcResultBuilder.<UpdateFlowOutput>failed()
+ .withRpcErrors(rpcResult.getErrors()).build());
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ LOG.error("Service call for updating flow with id {} failed for node {}",
+ getFlowId(updateFlowInput.getFlowRef()), nodeId, throwable);
+ resultFuture.setException(throwable);
+ }
+ }
+}