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.ChannelInitializer;
20 import io.netty.channel.ChannelOption;
21 import io.netty.channel.socket.SocketChannel;
22 import io.netty.handler.timeout.IdleStateHandler;
23 import io.netty.util.concurrent.Future;
24 import java.io.IOException;
25 import java.net.InetSocketAddress;
26 import java.time.Duration;
27 import java.util.concurrent.ScheduledExecutorService;
28 import java.util.concurrent.TimeUnit;
29 import org.checkerframework.checker.lock.qual.GuardedBy;
30 import org.checkerframework.checker.lock.qual.Holding;
31 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
32 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
33 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
34 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
35 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
36 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
37 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
38 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;
39 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
43 final class SinkSingletonService extends ChannelInitializer<SocketChannel> implements ClusterSingletonService {
44 private static final Logger LOG = LoggerFactory.getLogger(SinkSingletonService.class);
45 private static final ServiceGroupIdentifier SGID =
46 ServiceGroupIdentifier.create(SinkSingletonService.class.getName());
47 // TODO: allow different trees?
48 private static final DOMDataTreeIdentifier TREE = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
49 YangInstanceIdentifier.empty());
50 private static final ByteBuf TREE_REQUEST;
54 TREE_REQUEST = Unpooled.unreleasableBuffer(requestTree(TREE));
55 } catch (IOException e) {
56 throw new ExceptionInInitializerError(e);
60 private final BootstrapSupport bootstrapSupport;
61 private final DOMDataBroker dataBroker;
62 private final InetSocketAddress sourceAddress;
63 private final Duration reconnectDelay;
64 private final Duration keepaliveInterval;
67 private ChannelFuture futureChannel;
69 SinkSingletonService(final BootstrapSupport bootstrapSupport, final DOMDataBroker dataBroker,
70 final InetSocketAddress sourceAddress, final Duration reconnectDelay, final Duration keepaliveInterval) {
71 this.bootstrapSupport = requireNonNull(bootstrapSupport);
72 this.dataBroker = requireNonNull(dataBroker);
73 this.sourceAddress = requireNonNull(sourceAddress);
74 this.reconnectDelay = requireNonNull(reconnectDelay);
75 this.keepaliveInterval = requireNonNull(keepaliveInterval);
76 LOG.info("Replication sink from {} waiting for cluster-wide mastership", sourceAddress);
80 public ServiceGroupIdentifier getIdentifier() {
85 public synchronized void instantiateServiceInstance() {
86 LOG.info("Replication sink started with source {}", sourceAddress);
91 private void doConnect() {
92 LOG.info("Connecting to Source");
93 final Bootstrap bs = bootstrapSupport.newBootstrap();
94 final ScheduledExecutorService group = bs.config().group();
97 .option(ChannelOption.SO_KEEPALIVE, true)
99 .connect(sourceAddress, null);
101 futureChannel.addListener(compl -> channelResolved(compl, group));
105 public synchronized ListenableFuture<?> closeServiceInstance() {
106 if (futureChannel == null) {
107 return FluentFutures.immediateNullFluentFuture();
110 // FIXME: this is not really immediate. We also should be closing the resulting channel
111 return FluentFutures.immediateBooleanFluentFuture(futureChannel.cancel(true));
114 private synchronized void reconnect() {
119 protected void initChannel(final SocketChannel ch) {
121 .addLast("idleStateHandler", new IdleStateHandler(keepaliveInterval.toNanos(), 0, 0, TimeUnit.NANOSECONDS))
122 .addLast("keepaliveHandler", new KeepaliveHandler(this::reconnect))
123 .addLast("frameDecoder", new MessageFrameDecoder())
124 .addLast("requestHandler", new SinkRequestHandler(TREE, dataBroker.createMergingTransactionChain(
125 new SinkTransactionChainListener(ch))))
126 .addLast("frameEncoder", MessageFrameEncoder.INSTANCE);
129 private synchronized void channelResolved(final Future<?> completedFuture, final ScheduledExecutorService group) {
130 final Throwable cause = completedFuture.cause();
132 LOG.info("Failed to connect to source {}, reconnecting in {}", sourceAddress, reconnectDelay, cause);
133 group.schedule(() -> {
135 }, reconnectDelay.toNanos(), TimeUnit.NANOSECONDS);
139 final Channel ch = futureChannel.channel();
140 LOG.info("Channel {} established", ch);
141 ch.writeAndFlush(TREE_REQUEST);
144 private static ByteBuf requestTree(final DOMDataTreeIdentifier tree) throws IOException {
145 final ByteBuf ret = Unpooled.buffer();
147 try (ByteBufOutputStream stream = new ByteBufOutputStream(ret)) {
148 stream.writeByte(Constants.MSG_SUBSCRIBE_REQ);
149 try (NormalizedNodeDataOutput output = NormalizedNodeStreamVersion.current().newDataOutput(stream)) {
150 tree.getDatastoreType().writeTo(output);
151 output.writeYangInstanceIdentifier(tree.getRootIdentifier());