replicate/mdsal-replicate-netty/src/main/java/org/opendaylight/mdsal/replicate/netty/SinkSingletonService.java

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.mdsal.replicate.netty;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  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;
  42
  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;
  51
  52     static {
  53         try {
  54             TREE_REQUEST = Unpooled.unreleasableBuffer(requestTree(TREE));
  55         } catch (IOException e) {
  56             throw new ExceptionInInitializerError(e);
  57         }
  58     }
  59
  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;
  65
  66     @GuardedBy("this")
  67     private ChannelFuture futureChannel;
  68
  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);
  77     }
  78
  79     @Override
  80     public ServiceGroupIdentifier getIdentifier() {
  81         return SGID;
  82     }
  83
  84     @Override
  85     public synchronized void instantiateServiceInstance() {
  86         LOG.info("Replication sink started with source {}", sourceAddress);
  87         doConnect();
  88     }
  89
  90     @Holding("this")
  91     private void doConnect() {
  92         LOG.info("Connecting to Source");
  93         final Bootstrap bs = bootstrapSupport.newBootstrap();
  94         final ScheduledExecutorService group = bs.config().group();
  95
  96         futureChannel = bs
  97             .option(ChannelOption.SO_KEEPALIVE, true)
  98             .handler(this)
  99             .connect(sourceAddress, null);
 100
 101         futureChannel.addListener(compl -> channelResolved(compl, group));
 102     }
 103
 104     @Override
 105     public synchronized ListenableFuture<?> closeServiceInstance() {
 106         if (futureChannel == null) {
 107             return FluentFutures.immediateNullFluentFuture();
 108         }
 109
 110         // FIXME: this is not really immediate. We also should be closing the resulting channel
 111         return FluentFutures.immediateBooleanFluentFuture(futureChannel.cancel(true));
 112     }
 113
 114     private synchronized void reconnect() {
 115         doConnect();
 116     }
 117
 118     @Override
 119     protected void initChannel(final SocketChannel ch) {
 120         ch.pipeline()
 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);
 127     }
 128
 129     private synchronized void channelResolved(final Future<?> completedFuture, final ScheduledExecutorService group) {
 130         final Throwable cause = completedFuture.cause();
 131         if (cause != null) {
 132             LOG.info("Failed to connect to source {}, reconnecting in {}", sourceAddress, reconnectDelay, cause);
 133             group.schedule(() -> {
 134                 reconnect();
 135             }, reconnectDelay.toNanos(), TimeUnit.NANOSECONDS);
 136             return;
 137         }
 138
 139         final Channel ch = futureChannel.channel();
 140         LOG.info("Channel {} established", ch);
 141         ch.writeAndFlush(TREE_REQUEST);
 142     }
 143
 144     private static ByteBuf requestTree(final DOMDataTreeIdentifier tree) throws IOException {
 145         final ByteBuf ret = Unpooled.buffer();
 146
 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());
 152             }
 153         }
 154
 155         return ret;
 156     }
 157 }