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.util.concurrent.Future;
  23 import java.io.IOException;
  24 import java.net.InetSocketAddress;
  25 import java.time.Duration;
  26 import java.util.concurrent.ScheduledExecutorService;
  27 import java.util.concurrent.TimeUnit;
  28 import org.checkerframework.checker.lock.qual.GuardedBy;
  29 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  30 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  31 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  32 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  33 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  34 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
  35 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  36 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;
  37 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41 final class SinkSingletonService extends ChannelInitializer<SocketChannel> implements ClusterSingletonService {
  42     private static final Logger LOG = LoggerFactory.getLogger(SinkSingletonService.class);
  43     private static final ServiceGroupIdentifier SGID =
  44             ServiceGroupIdentifier.create(SinkSingletonService.class.getName());
  45     // TODO: allow different trees?
  46     private static final DOMDataTreeIdentifier TREE = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
  47         YangInstanceIdentifier.empty());
  48     private static final ByteBuf TREE_REQUEST;
  49
  50     static {
  51         try {
  52             TREE_REQUEST = Unpooled.unreleasableBuffer(requestTree(TREE));
  53         } catch (IOException e) {
  54             throw new ExceptionInInitializerError(e);
  55         }
  56     }
  57
  58     private final BootstrapSupport bootstrapSupport;
  59     private final DOMDataBroker dataBroker;
  60     private final InetSocketAddress sourceAddress;
  61     private final Duration reconnectDelay;
  62
  63     @GuardedBy("this")
  64     private ChannelFuture futureChannel;
  65
  66     SinkSingletonService(final BootstrapSupport bootstrapSupport, final DOMDataBroker dataBroker,
  67             final InetSocketAddress sourceAddress, final Duration reconnectDelay) {
  68         this.bootstrapSupport = requireNonNull(bootstrapSupport);
  69         this.dataBroker = requireNonNull(dataBroker);
  70         this.sourceAddress = requireNonNull(sourceAddress);
  71         this.reconnectDelay = requireNonNull(reconnectDelay);
  72         LOG.info("Replication sink from {} waiting for cluster-wide mastership", sourceAddress);
  73     }
  74
  75     @Override
  76     public ServiceGroupIdentifier getIdentifier() {
  77         return SGID;
  78     }
  79
  80     @Override
  81     public synchronized void instantiateServiceInstance() {
  82         LOG.info("Replication sink started with source {}", sourceAddress);
  83
  84         final Bootstrap bs = bootstrapSupport.newBootstrap();
  85         final ScheduledExecutorService group = bs.config().group();
  86
  87         futureChannel = bs
  88                 .option(ChannelOption.SO_KEEPALIVE, true)
  89                 .handler(this)
  90                 .connect(sourceAddress, null);
  91
  92         futureChannel.addListener(compl -> channelResolved(compl, group));
  93     }
  94
  95     @Override
  96     public synchronized ListenableFuture<?> closeServiceInstance() {
  97         // FIXME: initiate orderly shutdown
  98         return FluentFutures.immediateNullFluentFuture();
  99     }
 100
 101     @Override
 102     protected void initChannel(final SocketChannel ch) {
 103         ch.pipeline()
 104             .addLast("frameDecoder", new MessageFrameDecoder())
 105             .addLast("requestHandler", new SinkRequestHandler(TREE, dataBroker.createMergingTransactionChain(
 106                 new SinkTransactionChainListener(ch))))
 107             .addLast("frameEncoder", MessageFrameEncoder.INSTANCE);
 108     }
 109
 110     private synchronized void channelResolved(final Future<?> completedFuture, final ScheduledExecutorService group) {
 111         final Throwable cause = completedFuture.cause();
 112         if (cause != null) {
 113             LOG.info("Failed to connect to source {}, reconnecting in {}", sourceAddress, reconnectDelay, cause);
 114             group.schedule(() -> {
 115                 // FIXME: perform reconnect
 116             }, reconnectDelay.toNanos(), TimeUnit.NANOSECONDS);
 117             return;
 118         }
 119
 120         final Channel ch = futureChannel.channel();
 121         LOG.info("Channel {} established", ch);
 122         ch.writeAndFlush(TREE_REQUEST);
 123     }
 124
 125     private static ByteBuf requestTree(final DOMDataTreeIdentifier tree) throws IOException {
 126         final ByteBuf ret = Unpooled.buffer();
 127
 128         try (ByteBufOutputStream stream = new ByteBufOutputStream(ret)) {
 129             stream.writeByte(Constants.MSG_SUBSCRIBE_REQ);
 130             try (NormalizedNodeDataOutput output = NormalizedNodeStreamVersion.current().newDataOutput(stream)) {
 131                 tree.getDatastoreType().writeTo(output);
 132                 output.writeYangInstanceIdentifier(tree.getRootIdentifier());
 133             }
 134         }
 135
 136         return ret;
 137     }
 138 }