X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=netconf%2Fsal-netconf-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fnetconf%2Fsal%2Fconnect%2Fnetconf%2Fsal%2FNetconfDeviceRpc.java;h=6679f06020aa06ad66faa189a0801f45b633897d;hb=refs%2Fchanges%2F23%2F103723%2F3;hp=d1282d09b45c58c82785e391a11059705c2c25da;hpb=5a68db96acc2e92c82cbaee71f09df1634b81d10;p=netconf.git diff --git a/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/NetconfDeviceRpc.java b/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/NetconfDeviceRpc.java index d1282d09b4..6679f06020 100644 --- a/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/NetconfDeviceRpc.java +++ b/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/NetconfDeviceRpc.java @@ -7,98 +7,81 @@ */ package org.opendaylight.netconf.sal.connect.netconf.sal; -import com.google.common.base.Function; +import static java.util.Objects.requireNonNull; + import com.google.common.collect.Collections2; -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.ListenableFuture; -import java.util.Collection; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcAvailabilityListener; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcException; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcIdentifier; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcService; -import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.SettableFuture; +import org.opendaylight.mdsal.dom.api.DOMRpcAvailabilityListener; +import org.opendaylight.mdsal.dom.api.DOMRpcIdentifier; +import org.opendaylight.mdsal.dom.api.DOMRpcImplementationNotAvailableException; +import org.opendaylight.mdsal.dom.api.DOMRpcResult; +import org.opendaylight.mdsal.dom.api.DefaultDOMRpcException; +import org.opendaylight.mdsal.dom.spi.DefaultDOMRpcResult; import org.opendaylight.netconf.api.NetconfMessage; -import org.opendaylight.netconf.sal.connect.api.MessageTransformer; import org.opendaylight.netconf.sal.connect.api.RemoteDeviceCommunicator; +import org.opendaylight.netconf.sal.connect.api.RemoteDeviceServices.Rpcs; +import org.opendaylight.netconf.sal.connect.api.RpcTransformer; import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.concepts.NoOpListenerRegistration; +import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.common.RpcResult; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.model.api.RpcDefinition; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import org.opendaylight.yangtools.yang.model.api.SchemaPath; +import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext; /** - * Invokes RPC by sending netconf message via listener. Also transforms result from NetconfMessage to CompositeNode. + * Invokes RPC by sending netconf message via listener. Also transforms result from NetconfMessage to + * {@link NormalizedNode}. */ -public final class NetconfDeviceRpc implements DOMRpcService { - - private static final Function RPC_TO_RPC_IDENTIFIER = new Function() { - @Override - public DOMRpcIdentifier apply(final RpcDefinition input) { - return DOMRpcIdentifier.create(input.getPath()); - } - }; - - private final RemoteDeviceCommunicator listener; - private final MessageTransformer transformer; - private final Collection availableRpcs; +public final class NetconfDeviceRpc implements Rpcs.Normalized { + private final RemoteDeviceCommunicator communicator; + private final RpcTransformer transformer; + private final EffectiveModelContext modelContext; - public NetconfDeviceRpc(final SchemaContext schemaContext, final RemoteDeviceCommunicator listener, final MessageTransformer transformer) { - this.listener = listener; + public NetconfDeviceRpc(final EffectiveModelContext modelContext, final RemoteDeviceCommunicator communicator, + final RpcTransformer transformer) { + this.modelContext = requireNonNull(modelContext); + this.communicator = communicator; this.transformer = transformer; - - availableRpcs = Collections2.transform(schemaContext.getOperations(), RPC_TO_RPC_IDENTIFIER); } - @Nonnull @Override - public CheckedFuture invokeRpc(@Nonnull final SchemaPath type, @Nullable final NormalizedNode input) { - final NetconfMessage message = transformer.toRpcRequest(type, input); - final ListenableFuture> delegateFutureWithPureResult = listener.sendRequest(message, type.getLastComponent()); + @SuppressWarnings("checkstyle:IllegalCatch") + public ListenableFuture invokeRpc(final QName type, final NormalizedNode input) { + final ListenableFuture> delegateFuture = communicator.sendRequest( + transformer.toRpcRequest(type, input), type); - final ListenableFuture transformed = Futures.transform(delegateFutureWithPureResult, new Function, DOMRpcResult>() { + final SettableFuture ret = SettableFuture.create(); + Futures.addCallback(delegateFuture, new FutureCallback>() { @Override - public DOMRpcResult apply(final RpcResult input) { - if (input.isSuccessful()) { - return transformer.toRpcResult(input.getResult(), type); - } else { - // TODO check whether the listener sets errors properly - return new DefaultDOMRpcResult(input.getErrors()); + public void onSuccess(final RpcResult result) { + try { + ret.set(result.isSuccessful() ? transformer.toRpcResult(result.getResult(), type) + : new DefaultDOMRpcResult(result.getErrors())); + } catch (Exception cause) { + ret.setException(new DefaultDOMRpcException( + "Unable to parse rpc reply. type: " + type + " input: " + input, cause)); } } - }); - return Futures.makeChecked(transformed, new Function() { - @Nullable @Override - public DOMRpcException apply(@Nullable final Exception e) { - // FIXME what other possible exceptions are there ? - return new DOMRpcImplementationNotAvailableException(e, "Unable to invoke rpc %s", type); + public void onFailure(final Throwable cause) { + ret.setException(new DOMRpcImplementationNotAvailableException(cause, "Unable to invoke rpc %s", type)); } - }); + + }, MoreExecutors.directExecutor()); + return ret; } - @Nonnull @Override - public ListenerRegistration registerRpcListener(@Nonnull final T listener) { - - listener.onRpcAvailable(availableRpcs); + public ListenerRegistration registerRpcListener(final T listener) { + listener.onRpcAvailable(Collections2.transform(modelContext.getOperations(), + input -> DOMRpcIdentifier.create(input.getQName()))); - return new ListenerRegistration() { - @Override - public void close() { - // NOOP, no rpcs appear and disappear in this implementation - } - - @Override - public T getInstance() { - return listener; - } - }; + // NOOP, no rpcs appear and disappear in this implementation + return NoOpListenerRegistration.of(listener); } }