import static com.google.common.base.Verify.verifyNotNull;
import static java.util.Objects.requireNonNull;
+import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.opendaylight.restconf.common.patch.PatchStatusContext;
import org.opendaylight.restconf.common.patch.PatchStatusEntity;
import org.opendaylight.restconf.nb.rfc8040.Insert;
+import org.opendaylight.restconf.server.spi.OperationInput;
+import org.opendaylight.restconf.server.spi.OperationOutput;
+import org.opendaylight.restconf.server.spi.RpcImplementation;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.with.defaults.rev110601.WithDefaultsMode;
import org.opendaylight.yangtools.yang.common.Empty;
import org.opendaylight.yangtools.yang.common.ErrorTag;
private static final Logger LOG = LoggerFactory.getLogger(RestconfStrategy.class);
private final @NonNull EffectiveModelContext modelContext;
- private final @Nullable DOMRpcService rpcService;
+ private final @NonNull ImmutableMap<QName, RpcImplementation> localRpcs;
+ private final DOMRpcService rpcService;
- RestconfStrategy(final EffectiveModelContext modelContext, final @Nullable DOMRpcService rpcService) {
+ RestconfStrategy(final EffectiveModelContext modelContext, final ImmutableMap<QName, RpcImplementation> localRpcs,
+ final @Nullable DOMRpcService rpcService) {
this.modelContext = requireNonNull(modelContext);
+ this.localRpcs = requireNonNull(localRpcs);
this.rpcService = rpcService;
}
@Override
public void onFailure(final Throwable cause) {
- future.setFailure(TransactionUtil.decodeException(cause, "MERGE", path));
+ future.setFailure(TransactionUtil.decodeException(cause, "MERGE", path, modelContext));
}
}, MoreExecutors.directExecutor());
}
* @param insert {@link Insert}
* @return A {@link CreateOrReplaceResult}
*/
- public final @NonNull CreateOrReplaceResult putData(final YangInstanceIdentifier path, final NormalizedNode data,
- final @Nullable Insert insert) {
+ public final RestconfFuture<CreateOrReplaceResult> putData(final YangInstanceIdentifier path,
+ final NormalizedNode data, final @Nullable Insert insert) {
final var exists = TransactionUtil.syncAccess(exists(path), path);
final ListenableFuture<? extends CommitInfo> commitFuture;
commitFuture = replaceAndCommit(prepareWriteExecution(), path, data);
}
- TransactionUtil.syncCommit(commitFuture, "PUT", path);
- return exists ? CreateOrReplaceResult.REPLACED : CreateOrReplaceResult.CREATED;
+ final var ret = new SettableRestconfFuture<CreateOrReplaceResult>();
+
+ Futures.addCallback(commitFuture, new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ ret.set(exists ? CreateOrReplaceResult.REPLACED : CreateOrReplaceResult.CREATED);
+ }
+
+ @Override
+ public void onFailure(final Throwable cause) {
+ ret.setFailure(TransactionUtil.decodeException(cause, "PUT", path, modelContext));
+ }
+ }, MoreExecutors.directExecutor());
+
+ return ret;
}
private ListenableFuture<? extends CommitInfo> insertAndCommitPut(final YangInstanceIdentifier path,
* @param path path
* @param data data
* @param insert {@link Insert}
+ * @return A {@link RestconfFuture}
*/
- public final void postData(final YangInstanceIdentifier path, final NormalizedNode data,
+ public final RestconfFuture<Empty> postData(final YangInstanceIdentifier path, final NormalizedNode data,
final @Nullable Insert insert) {
final ListenableFuture<? extends CommitInfo> future;
if (insert != null) {
} else {
future = createAndCommit(prepareWriteExecution(), path, data);
}
- TransactionUtil.syncCommit(future, "POST", path);
+ final var ret = new SettableRestconfFuture<Empty>();
+
+ Futures.addCallback(future, new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ ret.set(Empty.value());
+ }
+
+ @Override
+ public void onFailure(final Throwable cause) {
+ ret.setFailure(TransactionUtil.decodeException(cause, "POST", path, modelContext));
+ }
+ }, MoreExecutors.directExecutor());
+
+ return ret;
}
private ListenableFuture<? extends CommitInfo> insertAndCommitPost(final YangInstanceIdentifier path,
* @param patch Patch context to be processed
* @return {@link PatchStatusContext}
*/
- public final @NonNull PatchStatusContext patchData(final PatchContext patch) {
+ public final @NonNull RestconfFuture<PatchStatusContext> patchData(final PatchContext patch) {
final var editCollection = new ArrayList<PatchStatusEntity>();
final var tx = prepareWriteExecution();
}
}
- // if no errors then submit transaction, otherwise cancel
- final var patchId = patch.patchId();
- if (noError) {
- try {
- TransactionUtil.syncCommit(tx.commit(), "PATCH", null);
- } catch (RestconfDocumentedException e) {
+ final var ret = new SettableRestconfFuture<PatchStatusContext>();
+ // We have errors
+ if (!noError) {
+ tx.cancel();
+ ret.set(new PatchStatusContext(modelContext, patch.patchId(), List.copyOf(editCollection), false, null));
+ return ret;
+ }
+
+ Futures.addCallback(tx.commit(), new FutureCallback<CommitInfo>() {
+ @Override
+ public void onSuccess(final CommitInfo result) {
+ ret.set(new PatchStatusContext(modelContext, patch.patchId(), List.copyOf(editCollection), true, null));
+ }
+
+ @Override
+ public void onFailure(final Throwable cause) {
// if errors occurred during transaction commit then patch failed and global errors are reported
- return new PatchStatusContext(modelContext, patchId, List.copyOf(editCollection), false, e.getErrors());
+ ret.set(new PatchStatusContext(modelContext, patch.patchId(), List.copyOf(editCollection), false,
+ TransactionUtil.decodeException(cause, "PATCH", null, modelContext).getErrors()));
}
+ }, MoreExecutors.directExecutor());
- return new PatchStatusContext(modelContext, patchId, List.copyOf(editCollection), true, null);
- } else {
- tx.cancel();
- return new PatchStatusContext(modelContext, patchId, List.copyOf(editCollection), false, null);
- }
+ return ret;
}
private void insertWithPointPost(final RestconfTransaction tx, final YangInstanceIdentifier path,
y -> builder.addChild((T) prepareData(y.getValue(), stateMap.get(y.getKey()))));
}
- public @NonNull RestconfFuture<Optional<ContainerNode>> invokeRpc(final QName type, final ContainerNode input) {
- final var ret = new SettableRestconfFuture<Optional<ContainerNode>>();
-
- final var local = rpcService;
+ public @NonNull RestconfFuture<OperationOutput> invokeRpc(final URI restconfURI, final QName type,
+ final OperationInput input) {
+ final var local = localRpcs.get(type);
if (local != null) {
- Futures.addCallback(local.invokeRpc(requireNonNull(type), requireNonNull(input)),
- new FutureCallback<DOMRpcResult>() {
- @Override
- public void onSuccess(final DOMRpcResult response) {
- final var errors = response.errors();
- if (errors.isEmpty()) {
- ret.set(Optional.ofNullable(response.value()));
- } else {
- LOG.debug("RPC invocation reported {}", response.errors());
- ret.setFailure(new RestconfDocumentedException("RPC implementation reported errors", null,
- response.errors()));
- }
- }
-
- @Override
- public void onFailure(final Throwable cause) {
- LOG.debug("RPC invocation failed, cause");
- if (cause instanceof RestconfDocumentedException ex) {
- ret.setFailure(ex);
- } else {
- // TODO: YangNetconfErrorAware if we ever get into a broader invocation scope
- ret.setFailure(new RestconfDocumentedException(cause,
- new RestconfError(ErrorType.RPC, ErrorTag.OPERATION_FAILED, cause.getMessage())));
- }
- }
- }, MoreExecutors.directExecutor());
- } else {
+ return local.invoke(restconfURI, input);
+ }
+ if (rpcService == null) {
LOG.debug("RPC invocation is not available");
- ret.setFailure(new RestconfDocumentedException("RPC invocation is not available",
+ return RestconfFuture.failed(new RestconfDocumentedException("RPC invocation is not available",
ErrorType.PROTOCOL, ErrorTag.OPERATION_NOT_SUPPORTED));
}
+
+ final var ret = new SettableRestconfFuture<OperationOutput>();
+ Futures.addCallback(rpcService.invokeRpc(requireNonNull(type), input.input()),
+ new FutureCallback<DOMRpcResult>() {
+ @Override
+ public void onSuccess(final DOMRpcResult response) {
+ final var errors = response.errors();
+ if (errors.isEmpty()) {
+ ret.set(input.newOperationOutput(response.value()));
+ } else {
+ LOG.debug("RPC invocation reported {}", response.errors());
+ ret.setFailure(new RestconfDocumentedException("RPC implementation reported errors", null,
+ response.errors()));
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable cause) {
+ LOG.debug("RPC invocation failed, cause");
+ if (cause instanceof RestconfDocumentedException ex) {
+ ret.setFailure(ex);
+ } else {
+ // TODO: YangNetconfErrorAware if we ever get into a broader invocation scope
+ ret.setFailure(new RestconfDocumentedException(cause,
+ new RestconfError(ErrorType.RPC, ErrorTag.OPERATION_FAILED, cause.getMessage())));
+ }
+ }
+ }, MoreExecutors.directExecutor());
return ret;
}
}
Code Review / netconf.git / blobdiff