platform/platform-impl/src/org/jetbrains/io/ChannelRegistrar.java

   1 /*
   2  * Copyright 2000-2016 JetBrains s.r.o.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.jetbrains.io;
  17
  18 import com.intellij.openapi.diagnostic.Logger;
  19 import io.netty.channel.*;
  20 import io.netty.channel.group.ChannelGroup;
  21 import io.netty.channel.group.ChannelGroupFuture;
  22 import io.netty.channel.group.DefaultChannelGroup;
  23 import io.netty.util.concurrent.ImmediateEventExecutor;
  24 import org.jetbrains.annotations.NotNull;
  25
  26 import java.util.Arrays;
  27 import java.util.concurrent.Future;
  28 import java.util.concurrent.TimeUnit;
  29
  30 @ChannelHandler.Sharable
  31 public final class ChannelRegistrar extends ChannelInboundHandlerAdapter {
  32   private static final Logger LOG = Logger.getInstance(ChannelRegistrar.class);
  33
  34   private final ChannelGroup openChannels = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
  35   private boolean isEventLoopGroupOwner;
  36
  37   public boolean isEmpty() {
  38     return openChannels.isEmpty();
  39   }
  40
  41   public void add(@NotNull Channel serverChannel, boolean isOwnEventLoopGroup) {
  42     this.isEventLoopGroupOwner = isOwnEventLoopGroup;
  43     assert serverChannel instanceof ServerChannel;
  44     openChannels.add(serverChannel);
  45   }
  46
  47   @Override
  48   public void channelActive(ChannelHandlerContext context) throws Exception {
  49     // we don't need to remove channel on close - ChannelGroup do it
  50     openChannels.add(context.channel());
  51
  52     super.channelActive(context);
  53   }
  54
  55   @NotNull
  56   public Future<?> close() {
  57     return close(isEventLoopGroupOwner);
  58   }
  59
  60   @NotNull
  61   private Future<?> close(boolean shutdownEventLoopGroup) {
  62     EventLoopGroup eventLoopGroup = null;
  63     if (shutdownEventLoopGroup) {
  64       for (Channel channel : openChannels) {
  65         if (channel instanceof ServerChannel) {
  66           eventLoopGroup = channel.eventLoop().parent();
  67           break;
  68         }
  69       }
  70     }
  71
  72     Future<?> result;
  73     try {
  74       Object[] channels = openChannels.toArray(new Channel[]{});
  75       ChannelGroupFuture groupFuture = openChannels.close();
  76       // server channels are closed in first turn, so, small timeout is relatively ok
  77       if (!groupFuture.awaitUninterruptibly(10, TimeUnit.SECONDS)) {
  78         LOG.warn("Cannot close all channels for 10 seconds, channels: " + Arrays.toString(channels));
  79       }
  80       result = groupFuture;
  81     }
  82     finally {
  83       if (eventLoopGroup != null) {
  84         result = eventLoopGroup.shutdownGracefully(1, 2, TimeUnit.NANOSECONDS);
  85       }
  86     }
  87     return result;
  88   }
  89 }