2 * Copyright (c) 2020 PANTHEON.tech, s.r.o. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.mdsal.replicate.netty;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.util.concurrent.ListenableFuture;
13 import io.netty.bootstrap.Bootstrap;
14 import io.netty.buffer.ByteBuf;
15 import io.netty.buffer.ByteBufOutputStream;
16 import io.netty.buffer.Unpooled;
17 import io.netty.channel.Channel;
18 import io.netty.channel.ChannelFuture;
19 import io.netty.channel.ChannelOption;
20 import io.netty.util.concurrent.Future;
21 import java.io.IOException;
22 import java.net.InetSocketAddress;
23 import java.time.Duration;
24 import java.util.concurrent.ScheduledExecutorService;
25 import org.checkerframework.checker.lock.qual.GuardedBy;
26 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
27 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
28 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
29 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
30 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
31 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
32 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;
33 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
34 import org.slf4j.Logger;
35 import org.slf4j.LoggerFactory;
37 final class SinkSingletonService implements ClusterSingletonService {
38 private static final Logger LOG = LoggerFactory.getLogger(SinkSingletonService.class);
39 private static final ServiceGroupIdentifier SGID =
40 ServiceGroupIdentifier.create(SinkSingletonService.class.getName());
41 // TODO: allow different trees?
42 private static final DOMDataTreeIdentifier TREE = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
43 YangInstanceIdentifier.empty());
44 private static final ByteBuf TREE_REQUEST;
48 TREE_REQUEST = requestTree(TREE);
49 } catch (IOException e) {
50 throw new ExceptionInInitializerError(e);
54 private final BootstrapSupport bootstrapSupport;
55 private final DOMDataBroker dataBroker;
56 private final InetSocketAddress sourceAddress;
57 private final Duration reconnectDelay;
60 private ChannelFuture futureChannel;
62 SinkSingletonService(final BootstrapSupport bootstrapSupport, final DOMDataBroker dataBroker,
63 final InetSocketAddress sourceAddress, final Duration reconnectDelay) {
64 this.bootstrapSupport = requireNonNull(bootstrapSupport);
65 this.dataBroker = requireNonNull(dataBroker);
66 this.sourceAddress = requireNonNull(sourceAddress);
67 this.reconnectDelay = requireNonNull(reconnectDelay);
71 public ServiceGroupIdentifier getIdentifier() {
76 public synchronized void instantiateServiceInstance() {
77 LOG.info("Replication sink started with source {}", sourceAddress);
79 final Bootstrap bs = bootstrapSupport.newBootstrap();
80 final ScheduledExecutorService group = bs.config().group();
83 .option(ChannelOption.SO_KEEPALIVE, true)
84 .connect(sourceAddress, null);
86 futureChannel.addListener(compl -> channelResolved(compl, group));
90 public synchronized ListenableFuture<?> closeServiceInstance() {
91 // TODO Auto-generated method stub
95 private synchronized void channelResolved(final Future<?> completedFuture, final ScheduledExecutorService group) {
96 if (completedFuture != futureChannel) {
97 // Future changed, this callback is irrelevant
101 final Channel channel = futureChannel.channel();
103 .addLast("frameDecoder", new MessageFrameDecoder())
104 .addLast("requestHandler", new SinkRequestHandler(TREE, dataBroker.createMergingTransactionChain(
105 new SinkTransactionChainListener(channel))))
106 .addLast("frameEncoder", MessageFrameEncoder.instance());
108 channel.writeAndFlush(TREE_REQUEST);
111 private static ByteBuf requestTree(final DOMDataTreeIdentifier tree) throws IOException {
112 final ByteBuf ret = Unpooled.buffer();
114 try (ByteBufOutputStream stream = new ByteBufOutputStream(ret)) {
115 try (NormalizedNodeDataOutput output = NormalizedNodeStreamVersion.current().newDataOutput(stream)) {
116 tree.getDatastoreType().writeTo(output);
117 output.writeYangInstanceIdentifier(tree.getRootIdentifier());