diff --git a/.idea/libraries/Netty.xml b/.idea/libraries/Netty.xml index 7cbf980422e9..3fbc53150ef8 100644 --- a/.idea/libraries/Netty.xml +++ b/.idea/libraries/Netty.xml @@ -1,11 +1,11 @@ - + - + \ No newline at end of file diff --git a/build/scripts/libLicenses.gant b/build/scripts/libLicenses.gant index bd420611b97e..b4fd69a80d60 100644 --- a/build/scripts/libLicenses.gant +++ b/build/scripts/libLicenses.gant @@ -248,7 +248,7 @@ libraryLicense(name: "XML-RPC", libraryName: "XmlRPC", version: "2.0", license: libraryLicense(name: "XStream", version: "1.4.3", license: "BSD", url: "http://xstream.codehaus.org/", licenseUrl: "http://xstream.codehaus.org/license.html") libraryLicense(name: "YourKit Java Profiler", libraryName: "yjp-controller-api-redist.jar", version: "8.0.x", license: "link (commercial license)", url: "http://yourkit.com/", licenseUrl: "http://www.yourkit.com/purchase/license.html") libraryLicense(name: "protobuf", version: "2.5.0", license: "New BSD", url: "http://code.google.com/p/protobuf/", licenseUrl: "http://code.google.com/p/protobuf/source/browse/trunk/COPYING.txt?r=367") -libraryLicense(name: "Netty", libraryName: "Netty", version: "3.6.6", license: "Apache 2.0", url: "http://netty.io", licenseUrl: "http://www.apache.org/licenses/LICENSE-2.0") +libraryLicense(name: "Netty", libraryName: "Netty", version: "4.0.8-SNAPSHOT", license: "Apache 2.0", url: "http://netty.io", licenseUrl: "http://www.apache.org/licenses/LICENSE-2.0") libraryLicense(name: "Kryo", libraryName: "Kryo", version: "1.04", license: "New BSD License", url: "http://code.google.com/p/kryo/", licenseUrl: "http://www.opensource.org/licenses/bsd-license.php") libraryLicense(name: "Snappy-Java", libraryName: "Snappy-Java", version: "1.0.5", license: "Apache 2.0", url: "http://code.google.com/p/snappy-java/", licenseUrl: "http://www.apache.org/licenses/LICENSE-2.0") libraryLicense(name: "Cucumber-Java", libraryName: "cucumber-java", version: "1.0.14", license: "MIT License", url: "https://github.com/cucumber/cucumber-jvm/", licenseUrl: "http://www.opensource.org/licenses/mit-license.html") diff --git a/java/compiler/impl/src/com/intellij/compiler/impl/ProblemsViewImpl.java b/java/compiler/impl/src/com/intellij/compiler/impl/ProblemsViewImpl.java index 27f61c47ff24..810ce712a57b 100644 --- a/java/compiler/impl/src/com/intellij/compiler/impl/ProblemsViewImpl.java +++ b/java/compiler/impl/src/com/intellij/compiler/impl/ProblemsViewImpl.java @@ -47,7 +47,7 @@ public class ProblemsViewImpl extends ProblemsView{ private static final String PROBLEMS_TOOLWINDOW_ID = "Problems"; private final ProblemsViewPanel myPanel; - private final SequentialTaskExecutor myViewUpdater = new SequentialTaskExecutor(new PooledThreadExecutor()); + private final SequentialTaskExecutor myViewUpdater = new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE); public ProblemsViewImpl(final Project project, final ToolWindowManager wm) { super(project); diff --git a/java/compiler/impl/src/com/intellij/compiler/server/BuildManager.java b/java/compiler/impl/src/com/intellij/compiler/server/BuildManager.java index e885a1670f1c..1db0adad0c3d 100644 --- a/java/compiler/impl/src/com/intellij/compiler/server/BuildManager.java +++ b/java/compiler/impl/src/com/intellij/compiler/server/BuildManager.java @@ -80,20 +80,20 @@ import com.intellij.util.io.storage.HeavyProcessLatch; import com.intellij.util.messages.MessageBusConnection; import com.intellij.util.net.NetUtils; import gnu.trove.THashSet; -import org.jboss.netty.bootstrap.ServerBootstrap; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.channel.group.ChannelGroupFuture; -import org.jboss.netty.channel.group.DefaultChannelGroup; -import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; -import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.handler.codec.protobuf.ProtobufDecoder; +import io.netty.handler.codec.protobuf.ProtobufEncoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.jetbrains.annotations.TestOnly; import org.jetbrains.ide.PooledThreadExecutor; +import org.jetbrains.io.ChannelRegistrar; +import org.jetbrains.io.NettyUtil; import org.jetbrains.jps.api.*; import org.jetbrains.jps.cmdline.BuildMain; import org.jetbrains.jps.cmdline.ClasspathBootstrap; @@ -106,7 +106,6 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.charset.Charset; import java.util.*; -import java.util.concurrent.Executor; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -117,7 +116,8 @@ import static org.jetbrains.jps.api.CmdlineRemoteProto.Message.ControllerMessage * @author Eugene Zhuravlev * Date: 9/6/11 */ -public class BuildManager implements ApplicationComponent{ +public class + BuildManager implements ApplicationComponent{ public static final Key ALLOW_AUTOMAKE = Key.create("_allow_automake_when_process_is_active_"); private static final Key FORCE_MODEL_LOADING_PARAMETER = Key.create(BuildParametersKeys.FORCE_MODEL_LOADING); @@ -152,8 +152,7 @@ public class BuildManager implements ApplicationComponent{ private final Map myAutomakeFutures = new HashMap(); private final Map myBuildsInProgress = Collections.synchronizedMap(new HashMap()); private final BuildProcessClasspathManager myClasspathManager = new BuildProcessClasspathManager(); - private final Executor myPooledThreadExecutor = new PooledThreadExecutor(); - private final SequentialTaskExecutor myRequestsProcessor = new SequentialTaskExecutor(myPooledThreadExecutor); + private final SequentialTaskExecutor myRequestsProcessor = new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE); private final Map myProjectDataMap = Collections.synchronizedMap(new HashMap()); private final BuildManagerPeriodicTask myAutoMakeTask = new BuildManagerPeriodicTask() { @@ -206,7 +205,8 @@ public class BuildManager implements ApplicationComponent{ } }; - private final ChannelGroup myAllOpenChannels = new DefaultChannelGroup("build-manager"); + private final ChannelRegistrar myChannelRegistrar = new ChannelRegistrar(); + private final BuildMessageDispatcher myMessageDispatcher = new BuildMessageDispatcher(); private volatile int myListenPort = -1; @Nullable @@ -354,7 +354,7 @@ public class BuildManager implements ApplicationComponent{ final CmdlineRemoteProto.Message.ControllerMessage message = CmdlineRemoteProto.Message.ControllerMessage.newBuilder().setType( CmdlineRemoteProto.Message.ControllerMessage.Type.FS_EVENT).setFsEvent(data.createNextEvent()).build(); - Channels.write(channel, CmdlineProtoUtil.toMessage(sessionId, message)); + channel.writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, message)); } } } @@ -574,7 +574,7 @@ public class BuildManager implements ApplicationComponent{ synchronized (myProjectDataMap) { ProjectData data = myProjectDataMap.get(projectPath); if (data == null) { - data = new ProjectData(new SequentialTaskExecutor(myPooledThreadExecutor)); + data = new ProjectData(new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE)); myProjectDataMap.put(projectPath, data); } if (isRebuild) { @@ -951,46 +951,25 @@ public class BuildManager implements ApplicationComponent{ } public void stopListening() { - final ChannelGroupFuture closeFuture = myAllOpenChannels.close(); - closeFuture.awaitUninterruptibly(); + myChannelRegistrar.close(); } private int startListening() throws Exception { - final ChannelFactory channelFactory = new NioServerSocketChannelFactory(myPooledThreadExecutor, myPooledThreadExecutor, 1); - final SimpleChannelUpstreamHandler channelRegistrar = new SimpleChannelUpstreamHandler() { + final ServerBootstrap bootstrap = NettyUtil.nioServerBootstrap(new NioEventLoopGroup(1, PooledThreadExecutor.INSTANCE)); + bootstrap.childHandler(new ChannelInitializer() { @Override - public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - myAllOpenChannels.add(e.getChannel()); - super.channelOpen(ctx, e); + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(myChannelRegistrar, + new ProtobufVarint32FrameDecoder(), + new ProtobufDecoder(CmdlineRemoteProto.Message.getDefaultInstance()), + new ProtobufVarint32LengthFieldPrepender(), + new ProtobufEncoder(), + myMessageDispatcher); } - - @Override - public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - myAllOpenChannels.remove(e.getChannel()); - super.channelClosed(ctx, e); - } - }; - ChannelPipelineFactory pipelineFactory = new ChannelPipelineFactory() { - @Override - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline( - channelRegistrar, - new ProtobufVarint32FrameDecoder(), - new ProtobufDecoder(CmdlineRemoteProto.Message.getDefaultInstance()), - new ProtobufVarint32LengthFieldPrepender(), - new ProtobufEncoder(), - myMessageDispatcher - ); - } - }; - final ServerBootstrap bootstrap = new ServerBootstrap(channelFactory); - bootstrap.setPipelineFactory(pipelineFactory); - bootstrap.setOption("child.tcpNoDelay", true); - bootstrap.setOption("child.keepAlive", true); - final int listenPort = NetUtils.findAvailableSocketPort(); - final Channel serverChannel = bootstrap.bind(new InetSocketAddress(NetUtils.getLoopbackAddress(), listenPort)); - myAllOpenChannels.add(serverChannel); - return listenPort; + }); + Channel serverChannel = bootstrap.bind(NetUtils.getLoopbackAddress(), 0).syncUninterruptibly().channel(); + myChannelRegistrar.add(serverChannel); + return ((InetSocketAddress)serverChannel.localAddress()).getPort(); } @TestOnly diff --git a/java/compiler/impl/src/com/intellij/compiler/server/BuildMessageDispatcher.java b/java/compiler/impl/src/com/intellij/compiler/server/BuildMessageDispatcher.java index 0c49759f6405..135b15488165 100644 --- a/java/compiler/impl/src/com/intellij/compiler/server/BuildMessageDispatcher.java +++ b/java/compiler/impl/src/com/intellij/compiler/server/BuildMessageDispatcher.java @@ -15,10 +15,13 @@ */ package com.intellij.compiler.server; -import com.intellij.openapi.application.ApplicationManager; import com.intellij.openapi.diagnostic.Logger; import com.intellij.util.containers.ConcurrentHashSet; -import org.jboss.netty.channel.*; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.util.AttributeKey; import org.jetbrains.annotations.Nullable; import org.jetbrains.jps.api.CmdlineProtoUtil; import org.jetbrains.jps.api.CmdlineRemoteProto; @@ -32,8 +35,12 @@ import java.util.concurrent.ConcurrentHashMap; * @author Eugene Zhuravlev * Date: 4/25/12 */ -class BuildMessageDispatcher extends SimpleChannelHandler { +@ChannelHandler.Sharable +class BuildMessageDispatcher extends SimpleChannelInboundHandler { private static final Logger LOG = Logger.getInstance("#com.intellij.compiler.server.BuildMessageDispatcher"); + + private static final AttributeKey SESSION_DATA = new AttributeKey("BuildMessageDispatcher.sessionData"); + private final Map myMessageHandlers = new ConcurrentHashMap(); private final Set myCanceledSessions = new ConcurrentHashSet(); @@ -54,7 +61,7 @@ class BuildMessageDispatcher extends SimpleChannelHandler { if (myCanceledSessions.add(sessionId)) { final Channel channel = getConnectedChannel(sessionId); if (channel != null) { - Channels.write(channel, CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createCancelCommand())); + channel.writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createCancelCommand())); } } } @@ -62,7 +69,7 @@ class BuildMessageDispatcher extends SimpleChannelHandler { @Nullable public Channel getConnectedChannel(final UUID sessionId) { final Channel channel = getAssociatedChannel(sessionId); - return channel != null && channel.isConnected()? channel : null; + return channel != null && channel.isActive()? channel : null; } @Nullable @@ -73,10 +80,8 @@ class BuildMessageDispatcher extends SimpleChannelHandler { @Override - public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { - final CmdlineRemoteProto.Message message = (CmdlineRemoteProto.Message)e.getMessage(); - - SessionData sessionData = (SessionData)ctx.getAttachment(); + protected void channelRead0(ChannelHandlerContext context, CmdlineRemoteProto.Message message) throws Exception { + SessionData sessionData = context.attr(SESSION_DATA).get(); UUID sessionId; if (sessionData == null) { @@ -86,11 +91,11 @@ class BuildMessageDispatcher extends SimpleChannelHandler { sessionData = myMessageHandlers.get(sessionId); if (sessionData != null) { - sessionData.channel = ctx.getChannel(); - ctx.setAttachment(sessionData); + sessionData.channel = context.channel(); + context.attr(SESSION_DATA).set(sessionData); } if (myCanceledSessions.contains(sessionId)) { - Channels.write(ctx.getChannel(), CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createCancelCommand())); + context.channel().writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createCancelCommand())); } } else { @@ -118,14 +123,14 @@ class BuildMessageDispatcher extends SimpleChannelHandler { if (params != null) { handler.buildStarted(sessionId); sessionData.params = null; - Channels.write(ctx.getChannel(), CmdlineProtoUtil.toMessage(sessionId, params)); + context.writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, params)); } else { cancelSession(sessionId); } } else { - handler.handleBuildMessage(ctx.getChannel(), sessionId, builderMessage); + handler.handleBuildMessage(context.channel(), sessionId, builderMessage); } break; @@ -136,12 +141,12 @@ class BuildMessageDispatcher extends SimpleChannelHandler { } @Override - public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + public void channelInactive(ChannelHandlerContext context) throws Exception { try { - super.channelClosed(ctx, e); + super.channelInactive(context); } finally { - final SessionData sessionData = (SessionData)ctx.getAttachment(); + final SessionData sessionData = context.attr(SESSION_DATA).get(); if (sessionData != null) { final BuilderMessageHandler handler = unregisterBuildMessageHandler(sessionData.sessionId); if (handler != null) { @@ -153,31 +158,10 @@ class BuildMessageDispatcher extends SimpleChannelHandler { } @Override - public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { - final Throwable cause = e.getCause(); + public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception { if (cause != null) { LOG.info(cause); } - ctx.sendUpstream(e); - } - - @Override - public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - try { - super.channelDisconnected(ctx, e); - } - finally { - final SessionData sessionData = (SessionData)ctx.getAttachment(); - if (sessionData != null) { // this is the context corresponding to some session - final Channel channel = e.getChannel(); - ApplicationManager.getApplication().executeOnPooledThread(new Runnable() { - @Override - public void run() { - channel.close(); - } - }); - } - } } private static final class SessionData { diff --git a/java/compiler/impl/src/com/intellij/compiler/server/BuilderMessageHandler.java b/java/compiler/impl/src/com/intellij/compiler/server/BuilderMessageHandler.java index fadf63b893ed..4681f859639d 100644 --- a/java/compiler/impl/src/com/intellij/compiler/server/BuilderMessageHandler.java +++ b/java/compiler/impl/src/com/intellij/compiler/server/BuilderMessageHandler.java @@ -15,7 +15,7 @@ */ package com.intellij.compiler.server; -import org.jboss.netty.channel.Channel; +import io.netty.channel.Channel; import org.jetbrains.jps.api.CmdlineRemoteProto; import java.util.UUID; diff --git a/java/compiler/impl/src/com/intellij/compiler/server/DefaultMessageHandler.java b/java/compiler/impl/src/com/intellij/compiler/server/DefaultMessageHandler.java index 35091f75238e..3cb12602c149 100644 --- a/java/compiler/impl/src/com/intellij/compiler/server/DefaultMessageHandler.java +++ b/java/compiler/impl/src/com/intellij/compiler/server/DefaultMessageHandler.java @@ -16,7 +16,7 @@ package com.intellij.compiler.server; import com.intellij.compiler.make.CachingSearcher; -import com.intellij.lang.StdLanguages; +import com.intellij.lang.java.JavaLanguage; import com.intellij.openapi.application.ApplicationManager; import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.progress.ProcessCanceledException; @@ -30,8 +30,7 @@ import com.intellij.psi.search.*; import com.intellij.psi.util.PsiUtil; import com.intellij.util.cls.ClsUtil; import com.intellij.util.concurrency.SequentialTaskExecutor; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.Channels; +import io.netty.channel.Channel; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.jetbrains.ide.PooledThreadExecutor; @@ -50,7 +49,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { private final Project myProject; private int myConstantSearchesCount = 0; private final CachingSearcher mySearcher; - private final SequentialTaskExecutor myTaskExecutor = new SequentialTaskExecutor(new PooledThreadExecutor()); + private final SequentialTaskExecutor myTaskExecutor = new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE); protected DefaultMessageHandler(Project project) { myProject = project; @@ -102,6 +101,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { } else { ApplicationManager.getApplication().runReadAction(new Runnable() { + @Override public void run() { try { String qualifiedName = ownerClassName.replace('$', '.'); @@ -137,8 +137,8 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { continue; } foundAtLeastOne = true; - final boolean sucess = performChangedConstantSearch(aClass, changedField, accessFlags, accessChanged, affectedPaths); - if (!sucess) { + final boolean success = performChangedConstantSearch(aClass, changedField, accessFlags, accessChanged, affectedPaths); + if (!success) { isSuccess.set(Boolean.FALSE); break; } @@ -175,7 +175,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { builder.setIsSuccess(false); LOG.debug("Constant search task: unsuccessful"); } - Channels.write(channel, CmdlineProtoUtil.toMessage(sessionId, CmdlineRemoteProto.Message.ControllerMessage.newBuilder().setType( + channel.writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, CmdlineRemoteProto.Message.ControllerMessage.newBuilder().setType( CmdlineRemoteProto.Message.ControllerMessage.Type.CONSTANT_SEARCH_RESULT).setConstantSearchResult(builder.build()).build() )); } @@ -219,11 +219,11 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { //} } } - catch (PsiInvalidElementAccessException e) { + catch (PsiInvalidElementAccessException ignored) { LOG.debug("Constant search task: PIEAE thrown while searching of usages of changed constant"); return false; } - catch (ProcessCanceledException e) { + catch (ProcessCanceledException ignored) { LOG.debug("Constant search task: PCE thrown while searching of usages of changed constant"); return false; } @@ -262,7 +262,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { } return true; } - catch (PsiInvalidElementAccessException e) { + catch (PsiInvalidElementAccessException ignored) { result.set(Boolean.FALSE); LOG.debug("Constant search task: PIEAE thrown while searching of usages of removed constant"); return false; @@ -285,6 +285,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { private static boolean processIdentifiers(PsiSearchHelper helper, @NotNull final PsiElementProcessor processor, @NotNull final String identifier, @NotNull SearchScope searchScope, short searchContext) { TextOccurenceProcessor processor1 = new TextOccurenceProcessor() { + @Override public boolean execute(PsiElement element, int offsetInElement) { return !(element instanceof PsiIdentifier) || processor.execute((PsiIdentifier)element); } @@ -332,7 +333,7 @@ public abstract class DefaultMessageHandler implements BuilderMessageHandler { if (containingFile == null) { return null; } - return StdLanguages.JAVA.equals(containingFile.getLanguage())? psiClass : null; + return JavaLanguage.INSTANCE.equals(containingFile.getLanguage())? psiClass : null; } element = element.getParent(); } diff --git a/java/compiler/impl/src/com/intellij/compiler/server/MessageHandlerWrapper.java b/java/compiler/impl/src/com/intellij/compiler/server/MessageHandlerWrapper.java index a90ff9b4270d..68b5d5c9d5c4 100644 --- a/java/compiler/impl/src/com/intellij/compiler/server/MessageHandlerWrapper.java +++ b/java/compiler/impl/src/com/intellij/compiler/server/MessageHandlerWrapper.java @@ -1,6 +1,6 @@ package com.intellij.compiler.server; -import org.jboss.netty.channel.Channel; +import io.netty.channel.Channel; import org.jetbrains.jps.api.CmdlineRemoteProto; import java.util.UUID; diff --git a/jps/jps-builders/src/org/jetbrains/jps/client/ProtobufClientMessageHandler.java b/jps/jps-builders/src/org/jetbrains/jps/client/ProtobufClientMessageHandler.java index 1134d30e197e..d6b428d9bf6c 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/client/ProtobufClientMessageHandler.java +++ b/jps/jps-builders/src/org/jetbrains/jps/client/ProtobufClientMessageHandler.java @@ -16,12 +16,12 @@ package org.jetbrains.jps.client; import com.google.protobuf.MessageLite; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelStateEvent; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelHandler; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; import org.jetbrains.annotations.NotNull; import org.jetbrains.jps.api.RequestFuture; +import org.jetbrains.jps.javac.JavacRemoteProto; import java.util.ArrayList; import java.util.UUID; @@ -32,7 +32,8 @@ import java.util.concurrent.Executor; * @author Eugene Zhuravlev * Date: 1/22/12 */ -final class ProtobufClientMessageHandler extends SimpleChannelHandler { +@ChannelHandler.Sharable +final class ProtobufClientMessageHandler extends SimpleChannelInboundHandler { private final ConcurrentHashMap> myHandlers = new ConcurrentHashMap>(); @NotNull private final UUIDGetter myUuidGetter; @@ -45,8 +46,9 @@ final class ProtobufClientMessageHandler exte myAsyncExec = asyncExec; } - public final void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { - final UUID messageUUID = myUuidGetter.getSessionUUID(e); + @Override + public final void channelRead0(ChannelHandlerContext context, MessageLite message) throws Exception { + final UUID messageUUID = myUuidGetter.getSessionUUID((JavacRemoteProto.Message)message); final RequestFuture future = myHandlers.get(messageUUID); final T handler = future != null ? future.getMessageHandler() : null; if (handler == null) { @@ -55,7 +57,7 @@ final class ProtobufClientMessageHandler exte else { boolean terminateSession = false; try { - terminateSession = handler.handleMessage((MessageLite)e.getMessage()); + terminateSession = handler.handleMessage(message); } catch (Exception ex) { terminateSession = true; @@ -78,8 +80,9 @@ final class ProtobufClientMessageHandler exte try { handler.sessionTerminated(); } - catch (Throwable ignored) { - ignored.printStackTrace(); + catch (Throwable e) { + //noinspection CallToPrintStackTrace + e.printStackTrace(); } } } @@ -89,30 +92,26 @@ final class ProtobufClientMessageHandler exte } } - public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - try { - super.channelClosed(ctx, e); - } - finally { - for (UUID uuid : new ArrayList(myHandlers.keySet())) { - terminateSession(uuid); - } - } - } - @Override - public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + public void channelInactive(ChannelHandlerContext context) throws Exception { try { - super.channelDisconnected(ctx, e); + super.channelInactive(context); } finally { - // make sure the client is in disconnected state - myAsyncExec.execute(new Runnable() { - @Override - public void run() { - myClient.disconnect(); + try { + for (UUID uuid : new ArrayList(myHandlers.keySet())) { + terminateSession(uuid); } - }); + } + finally { + // make sure the client is in disconnected state + myAsyncExec.execute(new Runnable() { + @Override + public void run() { + myClient.disconnect(); + } + }); + } } } diff --git a/jps/jps-builders/src/org/jetbrains/jps/client/SimpleProtobufClient.java b/jps/jps-builders/src/org/jetbrains/jps/client/SimpleProtobufClient.java index 06f41756fd03..19246ee7171a 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/client/SimpleProtobufClient.java +++ b/jps/jps-builders/src/org/jetbrains/jps/client/SimpleProtobufClient.java @@ -17,17 +17,17 @@ package org.jetbrains.jps.client; import com.google.protobuf.MessageLite; import com.intellij.openapi.diagnostic.Logger; -import org.jboss.netty.bootstrap.ClientBootstrap; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; -import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.*; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.protobuf.ProtobufDecoder; +import io.netty.handler.codec.protobuf.ProtobufEncoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; import org.jetbrains.annotations.Nullable; import org.jetbrains.jps.api.RequestFuture; -import java.net.InetSocketAddress; import java.util.UUID; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicReference; @@ -39,28 +39,27 @@ import java.util.concurrent.atomic.AtomicReference; public class SimpleProtobufClient { private static final Logger LOG = Logger.getInstance("#org.jetbrains.jps.client.SimpleProtobufClient"); - private static enum State { + private enum State { DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING } private final AtomicReference myState = new AtomicReference(State.DISCONNECTED); - protected final ChannelPipelineFactory myPipelineFactory; - protected final ChannelFactory myChannelFactory; + protected final ChannelInitializer myChannelInitializer; + protected final EventLoopGroup myEventLoopGroup; protected volatile ChannelFuture myConnectFuture; private final ProtobufClientMessageHandler myMessageHandler; public SimpleProtobufClient(final MessageLite msgDefaultInstance, final Executor asyncExec, final UUIDGetter uuidGetter) { myMessageHandler = new ProtobufClientMessageHandler(uuidGetter, this, asyncExec); - myChannelFactory = new NioClientSocketChannelFactory(asyncExec, asyncExec, 1); - myPipelineFactory = new ChannelPipelineFactory() { - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline( - new ProtobufVarint32FrameDecoder(), - new ProtobufDecoder(msgDefaultInstance), - new ProtobufVarint32LengthFieldPrepender(), - new ProtobufEncoder(), - myMessageHandler - ); + myEventLoopGroup = new NioEventLoopGroup(1, asyncExec); + myChannelInitializer = new ChannelInitializer() { + @Override + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(new ProtobufVarint32FrameDecoder(), + new ProtobufDecoder(msgDefaultInstance), + new ProtobufVarint32LengthFieldPrepender(), + new ProtobufEncoder(), + myMessageHandler); } }; } @@ -76,15 +75,10 @@ public class SimpleProtobufClient { boolean success = false; try { - final ClientBootstrap bootstrap = new ClientBootstrap(myChannelFactory); - bootstrap.setPipelineFactory(myPipelineFactory); - bootstrap.setOption("tcpNoDelay", true); - bootstrap.setOption("keepAlive", true); - final ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port)); - future.awaitUninterruptibly(); - + final Bootstrap bootstrap = new Bootstrap().group(myEventLoopGroup).channel(NioSocketChannel.class).handler(myChannelInitializer); + bootstrap.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true); + final ChannelFuture future = bootstrap.connect(host, port).syncUninterruptibly(); success = future.isSuccess(); - if (success) { myConnectFuture = future; try { @@ -94,13 +88,6 @@ public class SimpleProtobufClient { LOG.error(e); } } - else { - final Throwable reason = future.getCause(); - if (reason != null) { - throw reason; - } - } - return success; } finally { @@ -129,7 +116,7 @@ public class SimpleProtobufClient { catch (Throwable e) { LOG.error(e); } - final ChannelFuture closeFuture = future.getChannel().close(); + final ChannelFuture closeFuture = future.channel().close(); closeFuture.awaitUninterruptibly(); } } @@ -154,9 +141,10 @@ public class SimpleProtobufClient { final RequestFuture requestFuture = new RequestFuture(responseHandler, messageId, cancelAction); myMessageHandler.registerFuture(messageId, requestFuture); final ChannelFuture connectFuture = myConnectFuture; - final Channel channel = connectFuture != null? connectFuture.getChannel() : null; - if (channel != null && channel.isConnected()) { - Channels.write(channel, message).addListener(new ChannelFutureListener() { + final Channel channel = connectFuture != null? connectFuture.channel() : null; + if (channel != null && channel.isActive()) { + channel.writeAndFlush(message).addListener(new ChannelFutureListener() { + @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { notifyTerminated(messageId, requestFuture, responseHandler); diff --git a/jps/jps-builders/src/org/jetbrains/jps/client/UUIDGetter.java b/jps/jps-builders/src/org/jetbrains/jps/client/UUIDGetter.java index 675a6296f0a7..cba6db4fe296 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/client/UUIDGetter.java +++ b/jps/jps-builders/src/org/jetbrains/jps/client/UUIDGetter.java @@ -15,8 +15,8 @@ */ package org.jetbrains.jps.client; -import org.jboss.netty.channel.MessageEvent; import org.jetbrains.annotations.NotNull; +import org.jetbrains.jps.javac.JavacRemoteProto; import java.util.UUID; @@ -25,5 +25,5 @@ import java.util.UUID; * Date: 1/22/12 */ public interface UUIDGetter { - @NotNull UUID getSessionUUID(@NotNull MessageEvent e); + @NotNull UUID getSessionUUID(@NotNull JavacRemoteProto.Message message); } diff --git a/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildMain.java b/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildMain.java index 0bc21076a6c9..0e47e0c4d792 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildMain.java +++ b/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildMain.java @@ -18,16 +18,17 @@ package org.jetbrains.jps.cmdline; import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.util.io.FileUtil; import com.intellij.openapi.util.text.StringUtil; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.*; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.handler.codec.protobuf.ProtobufDecoder; +import io.netty.handler.codec.protobuf.ProtobufEncoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; import org.apache.log4j.Level; import org.apache.log4j.LogManager; import org.apache.log4j.xml.DOMConfigurator; -import org.jboss.netty.bootstrap.ClientBootstrap; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; -import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; import org.jetbrains.annotations.NonNls; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -45,6 +46,7 @@ import java.util.UUID; * @author Eugene Zhuravlev * Date: 4/16/12 */ +@SuppressWarnings("UseOfSystemOutOrSystemErr") public class BuildMain { private static final String LOG_CONFIG_FILE_NAME = "build-log.xml"; private static final String LOG_FILE_NAME = "build.log"; @@ -56,42 +58,37 @@ public class BuildMain { LOG = Logger.getInstance("#org.jetbrains.jps.cmdline.BuildMain"); } - private static NioClientSocketChannelFactory ourChannelFactory; + private static NioEventLoopGroup ourEventLoopGroup; public static void main(String[] args){ System.out.println("Build process started. Classpath: " + System.getProperty("java.class.path")); final String host = args[0]; final int port = Integer.parseInt(args[1]); final UUID sessionId = UUID.fromString(args[2]); + @SuppressWarnings("ConstantConditions") final File systemDir = new File(FileUtil.toCanonicalPath(args[3])); Utils.setSystemRoot(systemDir); - ourChannelFactory = new NioClientSocketChannelFactory(SharedThreadPool.getInstance(), SharedThreadPool.getInstance(), 1); - final ClientBootstrap bootstrap = new ClientBootstrap(ourChannelFactory); - bootstrap.setPipelineFactory(new ChannelPipelineFactory() { - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline( - new ProtobufVarint32FrameDecoder(), - new ProtobufDecoder(CmdlineRemoteProto.Message.getDefaultInstance()), - new ProtobufVarint32LengthFieldPrepender(), - new ProtobufEncoder(), - new MyMessageHandler(sessionId) - ); + ourEventLoopGroup = new NioEventLoopGroup(1, SharedThreadPool.getInstance()); + final Bootstrap bootstrap = new Bootstrap().group(ourEventLoopGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer() { + @Override + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(new ProtobufVarint32FrameDecoder(), + new ProtobufDecoder(CmdlineRemoteProto.Message.getDefaultInstance()), + new ProtobufVarint32LengthFieldPrepender(), + new ProtobufEncoder(), + new MyMessageHandler(sessionId)); } - }); - bootstrap.setOption("tcpNoDelay", true); - bootstrap.setOption("keepAlive", true); - - final ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port)); - future.awaitUninterruptibly(); + }).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true); + final ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port)).awaitUninterruptibly(); final boolean success = future.isSuccess(); - if (success) { - Channels.write(future.getChannel(), CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createParamRequest())); + future.channel().writeAndFlush(CmdlineProtoUtil.toMessage(sessionId, CmdlineProtoUtil.createParamRequest())); } else { - final Throwable reason = future.getCause(); + @SuppressWarnings("ThrowableResultOfMethodCallIgnored") + final Throwable reason = future.cause(); System.err.println("Error connecting to " + host + ":" + port + "; reason: " + (reason != null? reason.getMessage() : "unknown")); if (reason != null) { reason.printStackTrace(System.err); @@ -101,7 +98,7 @@ public class BuildMain { } } - private static class MyMessageHandler extends SimpleChannelHandler { + private static class MyMessageHandler extends SimpleChannelInboundHandler { private final UUID mySessionId; private volatile BuildSession mySession; @@ -110,10 +107,9 @@ public class BuildMain { } @Override - public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception { - CmdlineRemoteProto.Message message = (CmdlineRemoteProto.Message)e.getMessage(); + public void channelRead0(final ChannelHandlerContext context, CmdlineRemoteProto.Message message) throws Exception { final CmdlineRemoteProto.Message.Type type = message.getType(); - final Channel channel = ctx.getChannel(); + final Channel channel = context.channel(); if (type == CmdlineRemoteProto.Message.Type.CONTROLLER_MESSAGE) { final CmdlineRemoteProto.Message.ControllerMessage controllerMessage = message.getControllerMessage(); @@ -125,7 +121,9 @@ public class BuildMain { final BuildSession session = new BuildSession(mySessionId, channel, controllerMessage.getParamsMessage(), delta); mySession = session; SharedThreadPool.getInstance().executeOnPooledThread(new Runnable() { + @Override public void run() { + //noinspection finally try { session.run(); } @@ -172,19 +170,22 @@ public class BuildMain { } } - Channels.write(channel, CmdlineProtoUtil.toMessage(mySessionId, CmdlineProtoUtil.createFailure("Unsupported message type: " + type.name(), null))); + channel.writeAndFlush( + CmdlineProtoUtil.toMessage(mySessionId, CmdlineProtoUtil.createFailure("Unsupported message type: " + type.name(), null))); } @Override - public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + public void channelInactive(ChannelHandlerContext context) throws Exception { try { - super.channelClosed(ctx, e); + super.channelInactive(context); } finally { new Thread("Shutdown thread") { + @Override public void run() { + //noinspection finally try { - ourChannelFactory.releaseExternalResources(); + ourEventLoopGroup.shutdownGracefully(); } finally { System.exit(0); @@ -193,16 +194,6 @@ public class BuildMain { }.start(); } } - - @Override - public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - try { - super.channelDisconnected(ctx, e); - } - finally { - ctx.getChannel().close(); - } - } } private static void initLoggers() { @@ -217,6 +208,7 @@ public class BuildMain { } catch (IOException e) { System.err.println("Failed to configure logging: "); + //noinspection UseOfSystemOutOrSystemErr e.printStackTrace(System.err); } @@ -278,6 +270,7 @@ public class BuildMain { private static void ensureLogConfigExists(final File logConfig) throws IOException { if (!logConfig.exists()) { FileUtil.createIfDoesntExist(logConfig); + @SuppressWarnings("IOResourceOpenedButNotSafelyClosed") final InputStream in = BuildMain.class.getResourceAsStream("/" + DEFAULT_LOGGER_CONFIG); if (in != null) { try { diff --git a/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildSession.java b/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildSession.java index b841e323d119..f6d27c8f4253 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildSession.java +++ b/jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildSession.java @@ -24,16 +24,15 @@ import com.intellij.openapi.util.io.FileUtil; import com.intellij.openapi.util.text.StringUtil; import com.intellij.util.concurrency.SequentialTaskExecutor; import com.intellij.util.io.DataOutputStream; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.Channels; +import io.netty.channel.Channel; import org.jetbrains.annotations.Nullable; +import org.jetbrains.jps.TimingLog; import org.jetbrains.jps.api.*; import org.jetbrains.jps.builders.*; import org.jetbrains.jps.builders.java.JavaModuleBuildTargetType; import org.jetbrains.jps.builders.java.dependencyView.Callbacks; import org.jetbrains.jps.incremental.MessageHandler; import org.jetbrains.jps.incremental.TargetTypeRegistry; -import org.jetbrains.jps.TimingLog; import org.jetbrains.jps.incremental.Utils; import org.jetbrains.jps.incremental.fs.BuildFSState; import org.jetbrains.jps.incremental.fs.FSState; @@ -99,6 +98,7 @@ final class BuildSession implements Runnable, CanceledStatus { myBuildRunner = new BuildRunner(loader, filePaths, builderParams); } + @Override public void run() { Throwable error = null; final Ref hasErrors = new Ref(false); @@ -111,6 +111,7 @@ final class BuildSession implements Runnable, CanceledStatus { } runBuild(new MessageHandler() { + @Override public void processMessage(BuildMessage buildMessage) { final CmdlineRemoteProto.Message.BuilderMessage response; if (buildMessage instanceof FileGeneratedEvent) { @@ -151,7 +152,7 @@ final class BuildSession implements Runnable, CanceledStatus { response = null; } if (response != null) { - Channels.write(myChannel, CmdlineProtoUtil.toMessage(mySessionId, response)); + myChannel.writeAndFlush(CmdlineProtoUtil.toMessage(mySessionId, response)); } } }, this); @@ -441,9 +442,10 @@ final class BuildSession implements Runnable, CanceledStatus { private static void saveOnDisk(BufferExposingByteArrayOutputStream bytes, final File file) throws IOException { FileOutputStream fos = null; try { + //noinspection IOResourceOpenedButNotSafelyClosed fos = new FileOutputStream(file); } - catch (FileNotFoundException e) { + catch (FileNotFoundException ignored) { FileUtil.createIfDoesntExist(file); } @@ -541,7 +543,7 @@ final class BuildSession implements Runnable, CanceledStatus { } finally { try { - Channels.write(myChannel, lastMessage).await(); + myChannel.writeAndFlush(lastMessage).await(); } catch (InterruptedException e) { LOG.info(e); @@ -607,11 +609,8 @@ final class BuildSession implements Runnable, CanceledStatus { if (prev != null) { prev.setDone(); } - Channels.write(myChannel, - CmdlineProtoUtil.toMessage( - mySessionId, CmdlineRemoteProto.Message.BuilderMessage.newBuilder().setType(CmdlineRemoteProto.Message.BuilderMessage.Type.CONSTANT_SEARCH_TASK).setConstantSearchTask(task.build()).build() - ) - ); + myChannel.writeAndFlush(CmdlineProtoUtil.toMessage(mySessionId, CmdlineRemoteProto.Message.BuilderMessage.newBuilder() + .setType(CmdlineRemoteProto.Message.BuilderMessage.Type.CONSTANT_SEARCH_TASK).setConstantSearchTask(task.build()).build())); return future; } } diff --git a/jps/jps-builders/src/org/jetbrains/jps/cmdline/ClasspathBootstrap.java b/jps/jps-builders/src/org/jetbrains/jps/cmdline/ClasspathBootstrap.java index 6b9fef1d9e8f..ecf2ba05707a 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/cmdline/ClasspathBootstrap.java +++ b/jps/jps-builders/src/org/jetbrains/jps/cmdline/ClasspathBootstrap.java @@ -26,8 +26,8 @@ import com.intellij.util.PathUtilRt; import com.intellij.util.SystemProperties; import com.intellij.util.containers.ContainerUtil; import com.jgoodies.forms.layout.CellConstraints; +import io.netty.util.NetUtil; import net.n3.nanoxml.IXMLBuilder; -import org.jboss.netty.util.Version; import org.jetbrains.annotations.Nullable; import org.jetbrains.asm4.ClassVisitor; import org.jetbrains.asm4.ClassWriter; @@ -57,7 +57,7 @@ public class ClasspathBootstrap { @SuppressWarnings("unchecked") Class c = (Class)Class.forName(CLASS_NAME); aClass = c; } - catch (Throwable e) { + catch (Throwable ignored) { aClass = null; } managerClass = aClass; @@ -76,7 +76,7 @@ public class ClasspathBootstrap { @SuppressWarnings("unchecked") Class c = (Class)Class.forName(CLASS_NAME); aClass = c; } - catch (Throwable e) { + catch (Throwable ignored) { aClass = null; } managerClass = aClass; @@ -96,7 +96,7 @@ public class ClasspathBootstrap { cp.addAll(PathManager.getUtilClassPath()); // util cp.add(getResourcePath(Message.class)); // protobuf - cp.add(getResourcePath(Version.class)); // netty + cp.add(getResourcePath(NetUtil.class)); // netty cp.add(getResourcePath(ClassWriter.class)); // asm cp.add(getResourcePath(ClassVisitor.class)); // asm-commons cp.add(getResourcePath(JpsModel.class)); // jps-model-api @@ -141,7 +141,7 @@ public class ClasspathBootstrap { cp.add(getResourceFile(JpsModel.class)); // jps-model-api cp.add(getResourceFile(JpsModelImpl.class)); // jps-model-impl cp.add(getResourceFile(Message.class)); // protobuf - cp.add(getResourceFile(Version.class)); // netty + cp.add(getResourceFile(NetUtil.class)); // netty final Class optimizedFileManagerClass = getOptimizedFileManagerClass(); if (optimizedFileManagerClass != null) { diff --git a/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServer.java b/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServer.java index 76a6fe9b46d5..b901f95bb964 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServer.java +++ b/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServer.java @@ -15,74 +15,60 @@ */ package org.jetbrains.jps.javac; -import org.jboss.netty.bootstrap.ServerBootstrap; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.channel.group.ChannelGroupFuture; -import org.jboss.netty.channel.group.DefaultChannelGroup; -import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; -import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; -import org.jboss.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.*; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.ChannelGroupFuture; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.protobuf.ProtobufDecoder; +import io.netty.handler.codec.protobuf.ProtobufEncoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; +import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; +import io.netty.util.concurrent.ImmediateEventExecutor; import org.jetbrains.annotations.NotNull; import org.jetbrains.jps.api.CanceledStatus; import org.jetbrains.jps.service.SharedThreadPool; import javax.tools.*; import java.io.File; -import java.net.InetSocketAddress; import java.util.*; /** * @author Eugene Zhuravlev * Date: 1/22/12 */ +@SuppressWarnings("UseOfSystemOutOrSystemErr") public class JavacServer { public static final int DEFAULT_SERVER_PORT = 7878; public static final String SERVER_SUCCESS_START_MESSAGE = "Javac server started successfully. Listening on port: "; public static final String SERVER_ERROR_START_MESSAGE = "Error starting Javac Server: "; public static final String USE_ECLIPSE_COMPILER_PROPERTY = "use.eclipse.compiler"; - private final ChannelGroup myAllOpenChannels = new DefaultChannelGroup("javac-server"); - private final ChannelFactory myChannelFactory; - private final ChannelPipelineFactory myPipelineFactory; - - public JavacServer() { - myChannelFactory = new NioServerSocketChannelFactory(SharedThreadPool.getInstance(), SharedThreadPool.getInstance(), 1); - final ChannelRegistrar channelRegistrar = new ChannelRegistrar(); - final ChannelHandler compilationRequestsHandler = new CompilationRequestsHandler(); - myPipelineFactory = new ChannelPipelineFactory() { - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline( - channelRegistrar, - new ProtobufVarint32FrameDecoder(), - new ProtobufDecoder(JavacRemoteProto.Message.getDefaultInstance()), - new ProtobufVarint32LengthFieldPrepender(), - new ProtobufEncoder(), - compilationRequestsHandler - ); - } - }; - } + private ChannelRegistrar myChannelRegistrar; public void start(int listenPort) { - final ServerBootstrap bootstrap = new ServerBootstrap(myChannelFactory); - bootstrap.setPipelineFactory(myPipelineFactory); - bootstrap.setOption("child.tcpNoDelay", true); - bootstrap.setOption("child.keepAlive", true); - final Channel serverChannel = bootstrap.bind(new InetSocketAddress(listenPort)); - myAllOpenChannels.add(serverChannel); + final ServerBootstrap bootstrap = new ServerBootstrap().group(new NioEventLoopGroup(1, SharedThreadPool.getInstance())).channel(NioServerSocketChannel.class); + bootstrap.childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_KEEPALIVE, true); + myChannelRegistrar = new ChannelRegistrar(); + final ChannelHandler compilationRequestsHandler = new CompilationRequestsHandler(); + bootstrap.childHandler(new ChannelInitializer() { + @Override + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(myChannelRegistrar, + new ProtobufVarint32FrameDecoder(), + new ProtobufDecoder(JavacRemoteProto.Message.getDefaultInstance()), + new ProtobufVarint32LengthFieldPrepender(), + new ProtobufEncoder(), + compilationRequestsHandler); + } + }); + myChannelRegistrar.add(bootstrap.bind(listenPort).syncUninterruptibly().channel()); } public void stop() { - try { - final ChannelGroupFuture closeFuture = myAllOpenChannels.close(); - closeFuture.awaitUninterruptibly(); - } - finally { - myChannelFactory.releaseExternalResources(); - } + myChannelRegistrar.close().awaitUninterruptibly(); } public static void main(String[] args) { @@ -101,6 +87,7 @@ public class JavacServer { final JavacServer server = new JavacServer(); server.start(port); Runtime.getRuntime().addShutdownHook(new Thread("Shutdown hook thread") { + @Override public void run() { server.stop(); } @@ -116,8 +103,7 @@ public class JavacServer { } } - - public static JavacRemoteProto.Message compile(final ChannelHandlerContext ctx, + public static JavacRemoteProto.Message compile(final ChannelHandlerContext context, final UUID sessionId, List options, Collection files, @@ -129,28 +115,31 @@ public class JavacServer { final DiagnosticOutputConsumer diagnostic = new DiagnosticOutputConsumer() { @Override public void javaFileLoaded(File file) { - Channels.write(ctx.getChannel(), JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createSourceFileLoadedResponse(file))); + context.channel().writeAndFlush(JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createSourceFileLoadedResponse(file))); } + @Override public void outputLineAvailable(String line) { - Channels.write(ctx.getChannel(), JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createStdOutputResponse(line))); + context.channel().writeAndFlush(JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createStdOutputResponse(line))); } + @Override public void report(Diagnostic diagnostic) { final JavacRemoteProto.Message.Response response = JavacProtoUtil.createBuildMessageResponse(diagnostic); - Channels.write(ctx.getChannel(), JavacProtoUtil.toMessage(sessionId, response)); + context.channel().writeAndFlush(JavacProtoUtil.toMessage(sessionId, response)); } @Override public void registerImports(String className, Collection imports, Collection staticImports) { final JavacRemoteProto.Message.Response response = JavacProtoUtil.createClassDataResponse(className, imports, staticImports); - Channels.write(ctx.getChannel(), JavacProtoUtil.toMessage(sessionId, response)); + context.channel().writeAndFlush(JavacProtoUtil.toMessage(sessionId, response)); } }; final OutputFileConsumer outputSink = new OutputFileConsumer() { + @Override public void save(@NotNull OutputFileObject fileObject) { - Channels.write(ctx.getChannel(), JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createOutputObjectResponse(fileObject))); + context.channel().writeAndFlush(JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createOutputObjectResponse(fileObject))); } }; @@ -159,6 +148,7 @@ public class JavacServer { return JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createBuildCompletedResponse(rc)); } catch (Throwable e) { + //noinspection UseOfSystemOutOrSystemErr e.printStackTrace(System.err); return JavacProtoUtil.toMessage(sessionId, JavacProtoUtil.createFailure(e.getMessage(), e)); } @@ -182,18 +172,18 @@ public class JavacServer { return files; } - private class CompilationRequestsHandler extends SimpleChannelHandler { - - public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception { - final JavacRemoteProto.Message msg = (JavacRemoteProto.Message)e.getMessage(); - final UUID sessionId = JavacProtoUtil.fromProtoUUID(msg.getSessionId()); - final JavacRemoteProto.Message.Type messageType = msg.getMessageType(); + @ChannelHandler.Sharable + private class CompilationRequestsHandler extends SimpleChannelInboundHandler { + @Override + public void channelRead0(final ChannelHandlerContext context, JavacRemoteProto.Message message) throws Exception { + final UUID sessionId = JavacProtoUtil.fromProtoUUID(message.getSessionId()); + final JavacRemoteProto.Message.Type messageType = message.getMessageType(); JavacRemoteProto.Message reply = null; try { if (messageType == JavacRemoteProto.Message.Type.REQUEST) { - final JavacRemoteProto.Message.Request request = msg.getRequest(); + final JavacRemoteProto.Message.Request request = message.getRequest(); final JavacRemoteProto.Message.Request.Type requestType = request.getRequestType(); if (requestType == JavacRemoteProto.Message.Request.Type.COMPILE) { final List options = request.getOptionList(); @@ -214,11 +204,11 @@ public class JavacServer { final CancelHandler cancelHandler = new CancelHandler(); myCancelHandlers.add(cancelHandler); SharedThreadPool.getInstance().executeOnPooledThread(new Runnable() { + @Override public void run() { try { - final JavacRemoteProto.Message exitMsg = - compile(ctx, sessionId, options, files, cp, platformCp, srcPath, outs, cancelHandler); - Channels.write(ctx.getChannel(), exitMsg); + context.channel() + .writeAndFlush(compile(context, sessionId, options, files, cp, platformCp, srcPath, outs, cancelHandler)); } finally { myCancelHandlers.remove(cancelHandler); @@ -233,7 +223,9 @@ public class JavacServer { else if (requestType == JavacRemoteProto.Message.Request.Type.SHUTDOWN){ cancelBuilds(); new Thread("StopThread") { + @Override public void run() { + //noinspection finally try { JavacServer.this.stop(); } @@ -253,20 +245,51 @@ public class JavacServer { } finally { if (reply != null) { - Channels.write(ctx.getChannel(), reply); + context.channel().writeAndFlush(reply); } } } - - public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { - super.exceptionCaught(ctx, e); - } } - private class ChannelRegistrar extends SimpleChannelUpstreamHandler { - public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { - myAllOpenChannels.add(e.getChannel()); - super.channelOpen(ctx, e); + @ChannelHandler.Sharable + private static final class ChannelRegistrar extends ChannelInboundHandlerAdapter { + private final ChannelGroup openChannels = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE); + + public boolean isEmpty() { + return openChannels.isEmpty(); + } + + public void add(@NotNull Channel serverChannel) { + assert serverChannel instanceof ServerChannel; + openChannels.add(serverChannel); + } + + @Override + public void channelActive(ChannelHandlerContext context) throws Exception { + // we don't need to remove channel on close - ChannelGroup do it + openChannels.add(context.channel()); + + super.channelActive(context); + } + + public ChannelGroupFuture close() { + EventLoopGroup eventLoopGroup = null; + for (Channel channel : openChannels) { + if (channel instanceof ServerChannel) { + eventLoopGroup = channel.eventLoop().parent(); + break; + } + } + + ChannelGroupFuture future; + try { + future = openChannels.close(); + } + finally { + assert eventLoopGroup != null; + eventLoopGroup.shutdownGracefully(); + } + return future; } } @@ -280,6 +303,7 @@ public class JavacServer { myIsCanceled = true; } + @Override public boolean isCanceled() { return myIsCanceled; } diff --git a/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServerClient.java b/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServerClient.java index d9103ede38b3..bb553a9bac62 100644 --- a/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServerClient.java +++ b/jps/jps-builders/src/org/jetbrains/jps/javac/JavacServerClient.java @@ -15,7 +15,6 @@ */ package org.jetbrains.jps.javac; -import org.jboss.netty.channel.MessageEvent; import org.jetbrains.annotations.NotNull; import org.jetbrains.jps.api.RequestFuture; import org.jetbrains.jps.client.SimpleProtobufClient; @@ -33,9 +32,9 @@ public class JavacServerClient extends SimpleProtobufClient() { + @Override public void cancel(RequestFuture javacServerResponseHandlerRequestFuture) throws Exception { sendRequest(JavacProtoUtil.createCancelRequest(), null, null); } diff --git a/lib/netty-3.6.6.Final.jar b/lib/netty-3.6.6.Final.jar deleted file mode 100644 index 35cb0730cc14..000000000000 Binary files a/lib/netty-3.6.6.Final.jar and /dev/null differ diff --git a/lib/netty-all-4.0.7.Final.jar b/lib/netty-all-4.0.7.Final.jar new file mode 100644 index 000000000000..470f83b250e0 Binary files /dev/null and b/lib/netty-all-4.0.7.Final.jar differ diff --git a/lib/netty-all.jar b/lib/netty-all.jar new file mode 100644 index 000000000000..847efe5f5a99 Binary files /dev/null and b/lib/netty-all.jar differ diff --git a/lib/required_for_dist.txt b/lib/required_for_dist.txt index 5e07e02038a9..c5ea49e1dc4e 100644 --- a/lib/required_for_dist.txt +++ b/lib/required_for_dist.txt @@ -47,7 +47,7 @@ xmlrpc-2.0.jar xpp3-1.1.4-min.jar xstream-1.4.3.jar swingx-core-1.6.2.jar -netty-3.6.6.Final.jar +netty-all-4.0.7.Final.jar protobuf-2.5.0.jar rhino-js-1_7R4.jar proxy-vole_20120920.jar diff --git a/lib/src/netty-3.6.6.Final-sources.jar b/lib/src/netty-3.6.6.Final-sources.jar deleted file mode 100644 index 4144cf5d4da7..000000000000 Binary files a/lib/src/netty-3.6.6.Final-sources.jar and /dev/null differ diff --git a/lib/src/netty-all-4.0.7.Final-sources.jar b/lib/src/netty-all-4.0.7.Final-sources.jar new file mode 100644 index 000000000000..d5a47ae82d54 Binary files /dev/null and b/lib/src/netty-all-4.0.7.Final-sources.jar differ diff --git a/lib/src/netty-all-sources.jar b/lib/src/netty-all-sources.jar new file mode 100644 index 000000000000..07ac4063266a Binary files /dev/null and b/lib/src/netty-all-sources.jar differ diff --git a/platform/platform-api/src/com/intellij/ide/XmlRpcServer.java b/platform/platform-api/src/com/intellij/ide/XmlRpcServer.java index 24e06e662d74..16dafdcd8775 100644 --- a/platform/platform-api/src/com/intellij/ide/XmlRpcServer.java +++ b/platform/platform-api/src/com/intellij/ide/XmlRpcServer.java @@ -16,8 +16,8 @@ package com.intellij.ide; import com.intellij.openapi.components.ServiceManager; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.handler.codec.http.HttpRequest; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.FullHttpRequest; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -31,7 +31,7 @@ public interface XmlRpcServer { void removeHandler(String name); - boolean process(@NotNull String path, @NotNull HttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map handlers) throws IOException; + boolean process(@NotNull String path, @NotNull FullHttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map handlers) throws IOException; final class SERVICE { private SERVICE() { diff --git a/platform/platform-api/src/org/jetbrains/ide/HttpRequestHandler.java b/platform/platform-api/src/org/jetbrains/ide/HttpRequestHandler.java index aeccecedbe77..536858257296 100644 --- a/platform/platform-api/src/org/jetbrains/ide/HttpRequestHandler.java +++ b/platform/platform-api/src/org/jetbrains/ide/HttpRequestHandler.java @@ -15,18 +15,18 @@ */ package org.jetbrains.ide; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.QueryStringDecoder; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.QueryStringDecoder; import java.io.IOException; public abstract class HttpRequestHandler { - public boolean isSupported(HttpRequest request) { + public boolean isSupported(FullHttpRequest request) { return request.getMethod() == HttpMethod.GET || request.getMethod() == HttpMethod.HEAD; } - public abstract boolean process(QueryStringDecoder urlDecoder, HttpRequest request, ChannelHandlerContext context) + public abstract boolean process(QueryStringDecoder urlDecoder, FullHttpRequest request, ChannelHandlerContext context) throws IOException; } diff --git a/platform/platform-impl/src/com/intellij/ide/XmlRpcServerImpl.java b/platform/platform-impl/src/com/intellij/ide/XmlRpcServerImpl.java index f1c75e0e1e98..13ab6820f7f2 100644 --- a/platform/platform-impl/src/com/intellij/ide/XmlRpcServerImpl.java +++ b/platform/platform-impl/src/com/intellij/ide/XmlRpcServerImpl.java @@ -20,15 +20,14 @@ import com.intellij.openapi.extensions.Extensions; import com.intellij.openapi.util.text.StringUtil; import com.intellij.openapi.vfs.CharsetToolkit; import gnu.trove.THashMap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufInputStream; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.QueryStringDecoder; import org.apache.xmlrpc.*; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.buffer.ChannelBufferInputStream; -import org.jboss.netty.buffer.ChannelBuffers; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.HttpResponse; -import org.jboss.netty.handler.codec.http.QueryStringDecoder; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.jetbrains.ide.HttpRequestHandler; @@ -57,13 +56,13 @@ public class XmlRpcServerImpl implements XmlRpcServer { static final class XmlRpcRequestHandler extends HttpRequestHandler { @Override - public boolean isSupported(HttpRequest request) { + public boolean isSupported(FullHttpRequest request) { return request.getMethod() == HttpMethod.POST || request.getMethod() == HttpMethod.OPTIONS; } @Override - public boolean process(QueryStringDecoder urlDecoder, HttpRequest request, ChannelHandlerContext context) throws IOException { - return SERVICE.getInstance().process(urlDecoder.getPath(), request, context, null); + public boolean process(QueryStringDecoder urlDecoder, FullHttpRequest request, ChannelHandlerContext context) throws IOException { + return SERVICE.getInstance().process(urlDecoder.path(), request, context, null); } } @@ -83,14 +82,14 @@ public class XmlRpcServerImpl implements XmlRpcServer { } @Override - public boolean process(@NotNull String path, @NotNull HttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map handlers) throws IOException { + public boolean process(@NotNull String path, @NotNull FullHttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map handlers) throws IOException { if (!(path.isEmpty() || (path.length() == 1 && path.charAt(0) == '/') || path.equalsIgnoreCase("/RPC2"))) { return false; } if (request.getMethod() == HttpMethod.POST) { - ChannelBuffer result; - ChannelBufferInputStream in = new ChannelBufferInputStream(request.getContent()); + ByteBuf result; + ByteBufInputStream in = new ByteBufInputStream(request.content()); try { XmlRpcServerRequest xmlRpcServerRequest = new XmlRpcRequestProcessor().decodeRequest(in); @@ -100,10 +99,10 @@ public class XmlRpcServerImpl implements XmlRpcServer { } Object response = invokeHandler(getHandler(xmlRpcServerRequest.getMethodName(), handlers == null ? handlerMapping : handlers), xmlRpcServerRequest); - result = ChannelBuffers.copiedBuffer(new XmlRpcResponseProcessor().encodeResponse(response, CharsetToolkit.UTF8)); + result = Unpooled.copiedBuffer(new XmlRpcResponseProcessor().encodeResponse(response, CharsetToolkit.UTF8)); } catch (Throwable e) { - context.getChannel().close(); + context.channel().close(); LOG.error(e); return true; } @@ -111,12 +110,10 @@ public class XmlRpcServerImpl implements XmlRpcServer { in.close(); } - HttpResponse response = Responses.create("text/xml"); - response.setContent(result); - Responses.send(response, context.getChannel(), request); + Responses.send(Responses.create("text/xml", result), context.channel(), request); return true; } - else if (HttpMethod.POST.getName().equals(request.getHeader("Access-Control-Request-Method"))) { + else if (HttpMethod.POST.name().equals(request.headers().get("Access-Control-Request-Method"))) { LOG.assertTrue(request.getMethod() == HttpMethod.OPTIONS); Responses.sendOptionsResponse("POST, OPTIONS", request, context); return true; diff --git a/platform/platform-impl/src/org/jetbrains/ide/BuiltInServerManagerImpl.java b/platform/platform-impl/src/org/jetbrains/ide/BuiltInServerManagerImpl.java index 006f88b834cb..b783956b6bfb 100644 --- a/platform/platform-impl/src/org/jetbrains/ide/BuiltInServerManagerImpl.java +++ b/platform/platform-impl/src/org/jetbrains/ide/BuiltInServerManagerImpl.java @@ -14,7 +14,7 @@ import com.intellij.openapi.project.Project; import com.intellij.openapi.startup.StartupActivity; import com.intellij.openapi.util.Disposer; import com.intellij.openapi.util.ShutDownTracker; -import org.jboss.netty.channel.ChannelException; +import io.netty.channel.ChannelException; import org.jetbrains.annotations.NonNls; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -108,7 +108,8 @@ public class BuiltInServerManagerImpl extends BuiltInServerManager { } try { - server = new BuiltInServer(workerCount); + server = new BuiltInServer(); + detectedPortNumber = server.start(workerCount, defaultPort, PORTS_COUNT, true); } catch (ChannelException e) { LOG.info(e); @@ -121,6 +122,13 @@ public class BuiltInServerManagerImpl extends BuiltInServerManager { return; } + if (detectedPortNumber == -1) { + LOG.info("built-in server cannot be started, cannot bind to port"); + return; + } + + LOG.info("built-in server started, port " + detectedPortNumber); + Disposer.register(ApplicationManager.getApplication(), server); ShutDownTracker.getInstance().registerShutdownTask(new Runnable() { @Override @@ -131,14 +139,6 @@ public class BuiltInServerManagerImpl extends BuiltInServerManager { } } }); - - detectedPortNumber = server.start(defaultPort, PORTS_COUNT, true); - if (detectedPortNumber == -1) { - LOG.info("built-in server cannot be started, cannot bind to port"); - } - else { - LOG.info("built-in server started, port " + detectedPortNumber); - } } }); } diff --git a/platform/platform-impl/src/org/jetbrains/ide/PooledThreadExecutor.java b/platform/platform-impl/src/org/jetbrains/ide/PooledThreadExecutor.java index 1c241104cf0f..952393dd8fce 100644 --- a/platform/platform-impl/src/org/jetbrains/ide/PooledThreadExecutor.java +++ b/platform/platform-impl/src/org/jetbrains/ide/PooledThreadExecutor.java @@ -5,7 +5,12 @@ import org.jetbrains.annotations.NotNull; import java.util.concurrent.Executor; -public class PooledThreadExecutor implements Executor { +public final class PooledThreadExecutor implements Executor { + public static final PooledThreadExecutor INSTANCE = new PooledThreadExecutor(); + + private PooledThreadExecutor() { + } + @Override public void execute(@NotNull Runnable command) { ApplicationManager.getApplication().executeOnPooledThread(command); diff --git a/platform/platform-impl/src/org/jetbrains/io/BuiltInServer.java b/platform/platform-impl/src/org/jetbrains/io/BuiltInServer.java index 6f6130652bce..85d7089245ff 100644 --- a/platform/platform-impl/src/org/jetbrains/io/BuiltInServer.java +++ b/platform/platform-impl/src/org/jetbrains/io/BuiltInServer.java @@ -20,12 +20,12 @@ import com.intellij.openapi.Disposable; import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.util.Disposer; import com.intellij.util.net.NetUtils; -import org.jboss.netty.bootstrap.ServerBootstrap; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.channel.group.DefaultChannelGroup; -import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; -import org.jboss.netty.handler.codec.http.*; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.*; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.QueryStringDecoder; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.jetbrains.ide.CustomPortServerManager; @@ -35,73 +35,62 @@ import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.util.Map; -import java.util.concurrent.Executor; public class BuiltInServer implements Disposable { - private final ChannelGroup openChannels = new DefaultChannelGroup(); + private final ChannelRegistrar channelRegistrar = new ChannelRegistrar(); static final Logger LOG = Logger.getInstance(BuiltInServer.class); - private final NioServerSocketChannelFactory channelFactory; - - public BuiltInServer() { - this(1); - } - - public BuiltInServer(int workerCount) { - Executor pooledThreadExecutor = new PooledThreadExecutor(); - channelFactory = new NioServerSocketChannelFactory(pooledThreadExecutor, pooledThreadExecutor, workerCount); - } - public boolean isRunning() { - return !openChannels.isEmpty(); + return !channelRegistrar.isEmpty(); } public void start(int port) { - start(port, 1, false); + start(1, port, 1, false); } - public int start(int firstPort, int portsCount, boolean tryAnyPort) { + public int start(int workerCount, int firstPort, int portsCount, boolean tryAnyPort) { if (isRunning()) { throw new IllegalStateException("server already started"); } - ServerBootstrap bootstrap = createServerBootstrap(channelFactory, openChannels, null); + NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(workerCount, PooledThreadExecutor.INSTANCE); + ServerBootstrap bootstrap = createServerBootstrap(eventLoopGroup, channelRegistrar, null); int port = bind(firstPort, portsCount, tryAnyPort, bootstrap); - bindCustomPorts(firstPort, port); + bindCustomPorts(firstPort, port, eventLoopGroup); return port; } - static ServerBootstrap createServerBootstrap(NioServerSocketChannelFactory channelFactory, ChannelGroup openChannels, @Nullable Map xmlRpcHandlers) { - ServerBootstrap bootstrap = new ServerBootstrap(channelFactory); - bootstrap.setOption("child.tcpNoDelay", true); - bootstrap.setOption("child.keepAlive", true); + static ServerBootstrap createServerBootstrap(EventLoopGroup eventLoopGroup, final ChannelRegistrar channelRegistrar, @Nullable Map xmlRpcHandlers) { + ServerBootstrap bootstrap = NettyUtil.nioServerBootstrap(eventLoopGroup); if (xmlRpcHandlers == null) { - final ChannelHandler handler = new PortUnificationServerHandler(openChannels); - bootstrap.setPipelineFactory(new ChannelPipelineFactory() { + final PortUnificationServerHandler portUnificationServerHandler = new PortUnificationServerHandler(); + bootstrap.childHandler(new ChannelInitializer() { @Override - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline(handler); + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(channelRegistrar, portUnificationServerHandler); } }); } else { final XmlRpcDelegatingHttpRequestHandler handler = new XmlRpcDelegatingHttpRequestHandler(xmlRpcHandlers); - bootstrap.setPipelineFactory(new ChannelPipelineFactory() { + bootstrap.childHandler(new ChannelInitializer() { @Override - public ChannelPipeline getPipeline() throws Exception { - return Channels.pipeline(new HttpRequestDecoder(), new HttpChunkAggregator(1048576), new HttpResponseEncoder(), handler); + protected void initChannel(Channel channel) throws Exception { + channel.pipeline().addLast(channelRegistrar); + NettyUtil.initHttpHandlers(channel.pipeline()); + channel.pipeline().addLast(handler); } }); } return bootstrap; } - private void bindCustomPorts(int firstPort, int port) { + private void bindCustomPorts(int firstPort, int port, NioEventLoopGroup eventLoopGroup) { for (CustomPortServerManager customPortServerManager : CustomPortServerManager.EP_NAME.getExtensions()) { try { int customPortServerManagerPort = customPortServerManager.getPort(); - SubServer subServer = new SubServer(customPortServerManager, channelFactory); + SubServer subServer = new SubServer(customPortServerManager, eventLoopGroup); Disposer.register(this, subServer); if (customPortServerManagerPort != firstPort && customPortServerManagerPort != port) { subServer.bind(customPortServerManagerPort); @@ -117,55 +106,40 @@ public class BuiltInServer implements Disposable { InetAddress loopbackAddress = NetUtils.getLoopbackAddress(); for (int i = 0; i < portsCount; i++) { int port = firstPort + i; - try { - openChannels.add(bootstrap.bind(new InetSocketAddress(loopbackAddress, port))); + ChannelFuture future = bootstrap.bind(loopbackAddress, port).awaitUninterruptibly(); + if (future.isSuccess()) { + channelRegistrar.add(future.channel()); return port; } - catch (ChannelException e) { - if (!openChannels.isEmpty()) { - openChannels.close(); - openChannels.clear(); - } - - if (portsCount == 1) { - throw e; - } - else if (!tryAnyPort && i == (portsCount - 1)) { - LOG.error(e); - } + else if (!tryAnyPort && i == (portsCount - 1)) { + LOG.error(future.cause()); + return -1; } } - if (tryAnyPort) { - LOG.info("We cannot bind to our default range, so, try to bind to any free port"); - try { - Channel channel = bootstrap.bind(new InetSocketAddress(loopbackAddress, 0)); - openChannels.add(channel); - return ((InetSocketAddress)channel.getLocalAddress()).getPort(); - } - catch (ChannelException e) { - LOG.error(e); - } + LOG.info("We cannot bind to our default range, so, try to bind to any free port"); + ChannelFuture future = bootstrap.bind(loopbackAddress, 0).awaitUninterruptibly(); + if (future.isSuccess()) { + channelRegistrar.add(future.channel()); + return ((InetSocketAddress)future.channel().localAddress()).getPort(); + } + else { + LOG.error(future.cause()); + return -1; } - - return -1; } @Override public void dispose() { - try { - openChannels.close().awaitUninterruptibly(); - } - finally { - channelFactory.releaseExternalResources(); - } + channelRegistrar.close(); LOG.info("web server stopped"); } - public static void replaceDefaultHandler(@NotNull ChannelHandlerContext context, @NotNull SimpleChannelUpstreamHandler messageChannelHandler) { - context.getPipeline().replace(DelegatingHttpRequestHandler.class, "replacedDefaultHandler", messageChannelHandler); + public static void replaceDefaultHandler(@NotNull ChannelHandlerContext context, @NotNull SimpleChannelInboundHandler messageChannelHandler) { + context.pipeline().replace(DelegatingHttpRequestHandler.class, "replacedDefaultHandler", messageChannelHandler); } + @ChannelHandler.Sharable private static final class XmlRpcDelegatingHttpRequestHandler extends DelegatingHttpRequestHandlerBase { private final Map handlers; @@ -174,9 +148,9 @@ public class BuiltInServer implements Disposable { } @Override - protected boolean process(ChannelHandlerContext context, HttpRequest request, QueryStringDecoder urlDecoder) throws IOException { + protected boolean process(ChannelHandlerContext context, FullHttpRequest request, QueryStringDecoder urlDecoder) throws IOException { return (request.getMethod() == HttpMethod.POST || request.getMethod() == HttpMethod.OPTIONS) && - XmlRpcServer.SERVICE.getInstance().process(urlDecoder.getPath(), request, context, handlers); + XmlRpcServer.SERVICE.getInstance().process(urlDecoder.path(), request, context, handlers); } } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/ChannelExceptionHandler.java b/platform/platform-impl/src/org/jetbrains/io/ChannelExceptionHandler.java index 2ac00c1e9219..93def2e0194c 100644 --- a/platform/platform-impl/src/org/jetbrains/io/ChannelExceptionHandler.java +++ b/platform/platform-impl/src/org/jetbrains/io/ChannelExceptionHandler.java @@ -1,24 +1,33 @@ +/* + * Copyright 2000-2013 JetBrains s.r.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.jetbrains.io; import com.intellij.openapi.diagnostic.Logger; -import org.jboss.netty.channel.ChannelEvent; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelUpstreamHandler; -import org.jboss.netty.channel.ExceptionEvent; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; import java.net.ConnectException; -public final class ChannelExceptionHandler implements ChannelUpstreamHandler { +@ChannelHandler.Sharable +public final class ChannelExceptionHandler extends ChannelInboundHandlerAdapter { private static final Logger LOG = Logger.getInstance(ChannelExceptionHandler.class); @Override - public void handleUpstream(ChannelHandlerContext context, ChannelEvent event) throws Exception { - if (!(event instanceof ExceptionEvent)) { - context.sendUpstream(event); - return; - } - - Throwable cause = ((ExceptionEvent)event).getCause(); + public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception { // don't report about errors while connecting // WEB-7727 if (!(cause instanceof ConnectException) && !"Connection reset".equals(cause.getMessage())) { diff --git a/platform/platform-impl/src/org/jetbrains/io/ChannelRegistrar.java b/platform/platform-impl/src/org/jetbrains/io/ChannelRegistrar.java new file mode 100644 index 000000000000..96bb850d0899 --- /dev/null +++ b/platform/platform-impl/src/org/jetbrains/io/ChannelRegistrar.java @@ -0,0 +1,54 @@ +package org.jetbrains.io; + +import io.netty.channel.*; +import io.netty.channel.group.ChannelGroup; +import io.netty.channel.group.DefaultChannelGroup; +import io.netty.util.concurrent.ImmediateEventExecutor; +import org.jetbrains.annotations.NotNull; + +@ChannelHandler.Sharable +public final class ChannelRegistrar extends ChannelInboundHandlerAdapter { + private final ChannelGroup openChannels = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE); + + public boolean isEmpty() { + return openChannels.isEmpty(); + } + + public void add(@NotNull Channel serverChannel) { + assert serverChannel instanceof ServerChannel; + openChannels.add(serverChannel); + } + + @Override + public void channelActive(ChannelHandlerContext context) throws Exception { + // we don't need to remove channel on close - ChannelGroup do it + openChannels.add(context.channel()); + + super.channelActive(context); + } + + public void close() { + close(true); + } + + public void close(boolean shutdownEventLoopGroup) { + EventLoopGroup eventLoopGroup = null; + if (shutdownEventLoopGroup) { + for (Channel channel : openChannels) { + if (channel instanceof ServerChannel) { + eventLoopGroup = channel.eventLoop().parent(); + break; + } + } + } + + try { + openChannels.close().awaitUninterruptibly(); + } + finally { + if (eventLoopGroup != null) { + eventLoopGroup.shutdownGracefully(); + } + } + } +} \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/Decoder.java b/platform/platform-impl/src/org/jetbrains/io/Decoder.java index 45464e58860a..30052bde982f 100644 --- a/platform/platform-impl/src/org/jetbrains/io/Decoder.java +++ b/platform/platform-impl/src/org/jetbrains/io/Decoder.java @@ -15,24 +15,28 @@ */ package org.jetbrains.io; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.buffer.ChannelBuffers; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.SimpleChannelUpstreamHandler; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; import org.jetbrains.annotations.Nullable; -public abstract class Decoder extends SimpleChannelUpstreamHandler { - protected ChannelBuffer cumulation; +public abstract class Decoder extends SimpleChannelInboundHandler { + protected ByteBuf cumulation; + + protected Decoder() { + super(false); + } @Nullable - protected final ChannelBuffer getBufferIfSufficient(ChannelBuffer input, int requiredLength, ChannelHandlerContext context) { - if (!input.readable()) { + protected final ByteBuf getBufferIfSufficient(ByteBuf input, int requiredLength, ChannelHandlerContext context) { + if (!input.isReadable()) { return null; } if (cumulation == null) { if (input.readableBytes() < requiredLength) { - cumulation = context.getChannel().getConfig().getBufferFactory().getBuffer(requiredLength); + cumulation = context.channel().config().getAllocator().buffer(requiredLength); cumulation.writeBytes(input); return null; } @@ -46,7 +50,7 @@ public abstract class Decoder extends SimpleChannelUpstreamHandler { return null; } else { - ChannelBuffer buffer = ChannelBuffers.wrappedBuffer(cumulation, input); + ByteBuf buffer = Unpooled.wrappedBuffer(cumulation, input); input.skipBytes(input.readableBytes()); cumulation = null; return buffer; diff --git a/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandler.java b/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandler.java index e2464d3787da..a46c821719cc 100644 --- a/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandler.java +++ b/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandler.java @@ -18,13 +18,17 @@ package org.jetbrains.io; import com.intellij.openapi.application.ex.ApplicationInfoEx; import com.intellij.openapi.util.IconLoader; import com.intellij.util.ui.UIUtil; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponse; +import io.netty.handler.codec.http.QueryStringDecoder; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; import org.apache.sanselan.ImageFormat; import org.apache.sanselan.ImageWriteException; import org.apache.sanselan.Sanselan; -import org.jboss.netty.channel.ChannelHandler; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.QueryStringDecoder; import org.jetbrains.ide.BuiltInServerManager; import org.jetbrains.ide.HttpRequestHandler; @@ -34,23 +38,24 @@ import java.io.IOException; @ChannelHandler.Sharable final class DelegatingHttpRequestHandler extends DelegatingHttpRequestHandlerBase { + private static final AttributeKey PREV_HANDLER = new AttributeKey("DelegatingHttpRequestHandler.handler"); + @Override - protected boolean process(ChannelHandlerContext context, HttpRequest request, QueryStringDecoder urlDecoder) throws IOException, ImageWriteException { - HttpRequestHandler connectedHandler = (HttpRequestHandler)context.getAttachment(); + protected boolean process(ChannelHandlerContext context, FullHttpRequest request, QueryStringDecoder urlDecoder) throws IOException, ImageWriteException { + Attribute prevHandlerAttribute = context.attr(PREV_HANDLER); + HttpRequestHandler connectedHandler = prevHandlerAttribute.get(); if (connectedHandler != null) { if (connectedHandler.isSupported(request) && connectedHandler.process(urlDecoder, request, context)) { return true; } // prev cached connectedHandler is not suitable for this request, so, let's find it again - context.setAttachment(null); + prevHandlerAttribute.set(null); } for (HttpRequestHandler handler : BuiltInServerManager.EP_NAME.getExtensions()) { try { if (handler.isSupported(request) && handler.process(urlDecoder, request, context)) { - if (context.getAttachment() == null) { - context.setAttachment(handler); - } + prevHandlerAttribute.set(handler); return true; } } @@ -59,17 +64,26 @@ final class DelegatingHttpRequestHandler extends DelegatingHttpRequestHandlerBas } } - if (urlDecoder.getPath().equals("/favicon.ico")) { + if (urlDecoder.path().equals("/favicon.ico")) { Icon icon = IconLoader.findIcon(ApplicationInfoEx.getInstanceEx().getSmallIconUrl()); if (icon != null) { BufferedImage image = UIUtil.createImage(icon.getIconWidth(), icon.getIconHeight(), BufferedImage.TYPE_INT_ARGB); icon.paintIcon(null, image.getGraphics(), 0, 0); byte[] icoBytes = Sanselan.writeImageToBytes(image, ImageFormat.IMAGE_FORMAT_ICO, null); - Responses.send(icoBytes, FileResponses.createResponse(urlDecoder.getPath()), request, context); + HttpResponse response = Responses.create(FileResponses.getContentType(urlDecoder.path()), Unpooled.wrappedBuffer(icoBytes)); + Responses.addNoCache(response); + Responses.send(response, context.channel(), request); return true; } } return false; } + + @Override + public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception { + super.exceptionCaught(context, cause); + + context.attr(PREV_HANDLER).remove(); + } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandlerBase.java b/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandlerBase.java index 7fd497bb710c..8ee982469f3c 100644 --- a/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandlerBase.java +++ b/platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandlerBase.java @@ -15,43 +15,33 @@ */ package org.jetbrains.io; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ExceptionEvent; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelUpstreamHandler; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.QueryStringDecoder; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.QueryStringDecoder; -import static org.jboss.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND; - -abstract class DelegatingHttpRequestHandlerBase extends SimpleChannelUpstreamHandler { +abstract class DelegatingHttpRequestHandlerBase extends SimpleChannelInboundHandler { @Override - public final void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception { - if (!(event.getMessage() instanceof HttpRequest)) { - context.sendUpstream(event); - return; - } - - HttpRequest request = (HttpRequest)event.getMessage(); + protected void channelRead0(ChannelHandlerContext context, FullHttpRequest message) throws Exception { //if (BuiltInServer.LOG.isDebugEnabled()) { //BuiltInServer.LOG.debug(request.toString()); //} - if (!process(context, request, new QueryStringDecoder(request.getUri()))) { - Responses.sendStatus(request, context.getChannel(), NOT_FOUND); + if (!process(context, message, new QueryStringDecoder(message.getUri()))) { + Responses.sendStatus(message, context.channel(), HttpResponseStatus.NOT_FOUND); } } - protected abstract boolean process(ChannelHandlerContext context, HttpRequest request, QueryStringDecoder urlDecoder) throws Exception; + protected abstract boolean process(ChannelHandlerContext context, FullHttpRequest request, QueryStringDecoder urlDecoder) throws Exception; @Override - public final void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) throws Exception { + public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception { try { - NettyUtil.log(event.getCause(), BuiltInServer.LOG); + NettyUtil.log(cause, BuiltInServer.LOG); } finally { - context.setAttachment(null); - event.getChannel().close(); + context.channel().close(); } } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/FileResponses.java b/platform/platform-impl/src/org/jetbrains/io/FileResponses.java index ce70e3787d52..a29155202ab0 100644 --- a/platform/platform-impl/src/org/jetbrains/io/FileResponses.java +++ b/platform/platform-impl/src/org/jetbrains/io/FileResponses.java @@ -15,13 +15,14 @@ */ package org.jetbrains.io; -import org.jboss.netty.channel.*; -import org.jboss.netty.handler.codec.http.DefaultHttpResponse; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpRequest; -import org.jboss.netty.handler.codec.http.HttpResponse; -import org.jboss.netty.handler.ssl.SslHandler; -import org.jboss.netty.handler.stream.ChunkedFile; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.DefaultFileRegion; +import io.netty.handler.codec.http.*; +import io.netty.handler.ssl.SslHandler; +import io.netty.handler.stream.ChunkedFile; import javax.activation.MimetypesFileTypeMap; import java.io.File; @@ -30,28 +31,21 @@ import java.io.RandomAccessFile; import java.text.ParseException; import java.util.Date; -import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*; -import static org.jboss.netty.handler.codec.http.HttpHeaders.setContentLength; -import static org.jboss.netty.handler.codec.http.HttpResponseStatus.NOT_MODIFIED; -import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1; import static org.jetbrains.io.Responses.*; public class FileResponses { private static final MimetypesFileTypeMap FILE_MIMETYPE_MAP = new MimetypesFileTypeMap(); - public static HttpResponse createResponse(String path) { - HttpResponse response = create(FILE_MIMETYPE_MAP.getContentType(path)); - response.setHeader(CACHE_CONTROL, "no-cache, no-store, must-revalidate, max-age=0"); - response.setHeader(PRAGMA, "no-cache"); - return response; + public static String getContentType(String path) { + return FILE_MIMETYPE_MAP.getContentType(path); } private static boolean checkCache(HttpRequest request, Channel channel, long lastModified) { - String ifModifiedSince = request.getHeader(IF_MODIFIED_SINCE); + String ifModifiedSince = request.headers().get(HttpHeaders.Names.IF_MODIFIED_SINCE); if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) { try { if (Responses.DATE_FORMAT.get().parse(ifModifiedSince).getTime() >= lastModified) { - HttpResponse response = new DefaultHttpResponse(HTTP_1_1, NOT_MODIFIED); + HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_MODIFIED); addAllowAnyOrigin(response); addDate(response); addServer(response); @@ -76,30 +70,23 @@ public class FileResponses { RandomAccessFile raf = new RandomAccessFile(file, "r"); try { long fileLength = raf.length(); - HttpResponse response = createResponse(file.getPath()); + HttpResponse response = create(getContentType(file.getPath())); + addNoCache(response); addCommonHeaders(response); - response.setHeader(LAST_MODIFIED, Responses.DATE_FORMAT.get().format(new Date(file.lastModified()))); + response.headers().set(HttpHeaders.Names.LAST_MODIFIED, Responses.DATE_FORMAT.get().format(new Date(file.lastModified()))); boolean keepAlive = addKeepAliveIfNeed(response, request); if (request.getMethod() != HttpMethod.HEAD) { - setContentLength(response, fileLength); + HttpHeaders.setContentLength(response, fileLength); } ChannelFuture future = channel.write(response); - if (request.getMethod() != HttpMethod.HEAD) { - if (channel.getPipeline().get(SslHandler.class) == null) { - // No encryption - use zero-copy. - final FileRegion region = new DefaultFileRegion(raf.getChannel(), 0, fileLength); - future = channel.write(region); - future.addListener(new ChannelFutureListener() { - @Override - public void operationComplete(ChannelFuture future) { - region.releaseExternalResources(); - } - }); + if (channel.pipeline().get(SslHandler.class) == null) { + // No encryption - use zero-copy + future = channel.write(new DefaultFileRegion(raf.getChannel(), 0, fileLength)); } else { - // Cannot use zero-copy with HTTPS. + // Cannot use zero-copy with HTTPS future = channel.write(new ChunkedFile(raf, 0, fileLength, 8192)); } } @@ -107,6 +94,7 @@ public class FileResponses { if (!keepAlive) { future.addListener(ChannelFutureListener.CLOSE); } + channel.flush(); fileWillBeClosed = true; } @@ -116,4 +104,4 @@ public class FileResponses { } } } -} +} \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/NettyUtil.java b/platform/platform-impl/src/org/jetbrains/io/NettyUtil.java index 293a22b8e720..c37d7db6cd2e 100644 --- a/platform/platform-impl/src/org/jetbrains/io/NettyUtil.java +++ b/platform/platform-impl/src/org/jetbrains/io/NettyUtil.java @@ -17,12 +17,18 @@ package org.jetbrains.io; import com.intellij.openapi.diagnostic.Logger; import com.intellij.openapi.util.ActionCallback; -import org.jboss.netty.bootstrap.ClientBootstrap; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.ChannelException; -import org.jboss.netty.channel.ChannelFactory; -import org.jboss.netty.channel.ChannelFuture; +import io.netty.bootstrap.Bootstrap; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.*; +import io.netty.channel.oio.OioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.channel.socket.oio.OioSocketChannel; +import io.netty.handler.codec.http.HttpContentCompressor; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpRequestDecoder; +import io.netty.handler.codec.http.HttpResponseEncoder; import org.jetbrains.annotations.Nullable; +import org.jetbrains.ide.PooledThreadExecutor; import java.io.IOException; import java.net.SocketAddress; @@ -40,18 +46,18 @@ public final class NettyUtil { } } - public static Channel connectClient(ClientBootstrap bootstrap, SocketAddress remoteAddress, ActionCallback asyncResult) { + public static Channel connectClient(Bootstrap bootstrap, SocketAddress remoteAddress, ActionCallback asyncResult) { return connect(bootstrap, remoteAddress, asyncResult, DEFAULT_CONNECT_ATTEMPT_COUNT); } @Nullable - public static Channel connect(ClientBootstrap bootstrap, SocketAddress remoteAddress, ActionCallback asyncResult, int maxAttemptCount) { + public static Channel connect(Bootstrap bootstrap, SocketAddress remoteAddress, ActionCallback asyncResult, int maxAttemptCount) { int attemptCount = 0; while (true) { try { ChannelFuture future = bootstrap.connect(remoteAddress).await(); if (future.isSuccess()) { - return future.getChannel(); + return future.channel(); } else if (asyncResult.isRejected()) { return null; @@ -84,13 +90,29 @@ public final class NettyUtil { // applicable only in case of ClientBootstrap&OioClientSocketChannelFactory public static void closeAndReleaseFactory(Channel channel) { - ChannelFactory channelFactory = channel.getFactory(); + EventLoop channelFactory = channel.eventLoop(); try { channel.close().awaitUninterruptibly(); } finally { // in our case it does nothing, we don't use ExecutorService, but we are aware of future changes - channelFactory.releaseExternalResources(); + channelFactory.shutdownGracefully(); } } + + public static ServerBootstrap nioServerBootstrap(EventLoopGroup eventLoopGroup) { + ServerBootstrap bootstrap = new ServerBootstrap().group(eventLoopGroup).channel(NioServerSocketChannel.class); + bootstrap.childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_KEEPALIVE, true); + return bootstrap; + } + + public static Bootstrap oioClientBootstrap() { + Bootstrap bootstrap = new Bootstrap().group(new OioEventLoopGroup(1, PooledThreadExecutor.INSTANCE)).channel(OioSocketChannel.class); + bootstrap.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true); + return bootstrap; + } + + public static void initHttpHandlers(ChannelPipeline pipeline) { + pipeline.addLast(new HttpRequestDecoder(), new HttpObjectAggregator(1048576), new HttpResponseEncoder(), new HttpContentCompressor()); + } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/PortUnificationServerHandler.java b/platform/platform-impl/src/org/jetbrains/io/PortUnificationServerHandler.java index 1662ab01e16a..6f26fd4136ef 100644 --- a/platform/platform-impl/src/org/jetbrains/io/PortUnificationServerHandler.java +++ b/platform/platform-impl/src/org/jetbrains/io/PortUnificationServerHandler.java @@ -16,30 +16,25 @@ package org.jetbrains.io; import com.intellij.openapi.util.AtomicNotNullLazyValue; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.channel.*; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.handler.codec.compression.ZlibDecoder; -import org.jboss.netty.handler.codec.compression.ZlibEncoder; -import org.jboss.netty.handler.codec.compression.ZlibWrapper; -import org.jboss.netty.handler.codec.http.HttpChunkAggregator; -import org.jboss.netty.handler.codec.http.HttpContentCompressor; -import org.jboss.netty.handler.codec.http.HttpRequestDecoder; -import org.jboss.netty.handler.codec.http.HttpResponseEncoder; -import org.jboss.netty.handler.ssl.SslHandler; -import org.jboss.netty.handler.stream.ChunkedWriteHandler; +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPipeline; +import io.netty.handler.codec.compression.JZlibEncoder; +import io.netty.handler.codec.compression.JdkZlibDecoder; +import io.netty.handler.codec.compression.ZlibWrapper; +import io.netty.handler.ssl.SslHandler; +import io.netty.handler.stream.ChunkedWriteHandler; import org.jetbrains.annotations.NotNull; -import org.jetbrains.annotations.Nullable; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; -import java.net.SocketAddress; import java.security.KeyStore; import java.security.Security; @ChannelHandler.Sharable -final class PortUnificationServerHandler extends Decoder { +final class PortUnificationServerHandler extends Decoder { private static final AtomicNotNullLazyValue SSL_SERVER_CONTEXT = new AtomicNotNullLazyValue() { @NotNull @Override @@ -69,73 +64,58 @@ final class PortUnificationServerHandler extends Decoder { private final boolean detectSsl; private final boolean detectGzip; - private final ChannelGroup openChannels; private final DelegatingHttpRequestHandler delegatingHttpRequestHandler; - public PortUnificationServerHandler(ChannelGroup openChannels) { - this(new DelegatingHttpRequestHandler(), openChannels, true, true); + public PortUnificationServerHandler() { + this(new DelegatingHttpRequestHandler(), true, true); } - private PortUnificationServerHandler(DelegatingHttpRequestHandler delegatingHttpRequestHandler, @Nullable ChannelGroup openChannels, boolean detectSsl, boolean detectGzip) { + private PortUnificationServerHandler(DelegatingHttpRequestHandler delegatingHttpRequestHandler, boolean detectSsl, boolean detectGzip) { this.delegatingHttpRequestHandler = delegatingHttpRequestHandler; - this.openChannels = openChannels; this.detectSsl = detectSsl; this.detectGzip = detectGzip; } @Override - public void channelOpen(ChannelHandlerContext context, ChannelStateEvent e) { - if (openChannels != null) { - openChannels.add(e.getChannel()); + protected void channelRead0(ChannelHandlerContext context, ByteBuf message) throws Exception { + ByteBuf buffer = getBufferIfSufficient(message, 5, context); + if (buffer == null) { + message.release(); + } + else { + decode(context, buffer); } } - @Override - public void messageReceived(ChannelHandlerContext context, MessageEvent e) throws Exception { - Object m = e.getMessage(); - if (!(m instanceof ChannelBuffer)) { - context.sendUpstream(e); - return; - } - - ChannelBuffer buffer = getBufferIfSufficient((ChannelBuffer)m, 5, context); - if (buffer != null) { - decode(context, buffer, e.getRemoteAddress()); - } - } - - private void decode(ChannelHandlerContext context, ChannelBuffer buffer, SocketAddress remoteAddress) throws Exception { - ChannelPipeline pipeline = context.getPipeline(); + private void decode(ChannelHandlerContext context, ByteBuf buffer) throws Exception { + ChannelPipeline pipeline = context.pipeline(); if (detectSsl && SslHandler.isEncrypted(buffer)) { SSLEngine engine = SSL_SERVER_CONTEXT.getValue().createSSLEngine(); engine.setUseClientMode(false); pipeline.addLast("ssl", new SslHandler(engine)); pipeline.addLast("streamer", new ChunkedWriteHandler()); - pipeline.addLast("unificationWOSsl", new PortUnificationServerHandler(delegatingHttpRequestHandler, null, false, detectGzip)); + pipeline.addLast("unificationWOSsl", new PortUnificationServerHandler(delegatingHttpRequestHandler, false, detectGzip)); } else { int magic1 = buffer.getUnsignedByte(buffer.readerIndex()); int magic2 = buffer.getUnsignedByte(buffer.readerIndex() + 1); if (detectGzip && magic1 == 31 && magic2 == 139) { - pipeline.addLast("gzipDeflater", new ZlibEncoder(ZlibWrapper.GZIP)); - pipeline.addLast("gzipInflater", new ZlibDecoder(ZlibWrapper.GZIP)); - pipeline.addLast("unificationWOGzip", new PortUnificationServerHandler(delegatingHttpRequestHandler, null, detectSsl, false)); + pipeline.addLast("gzipDeflater", new JZlibEncoder(ZlibWrapper.GZIP)); + pipeline.addLast("gzipInflater", new JdkZlibDecoder(ZlibWrapper.GZIP)); + pipeline.addLast("unificationWOGzip", new PortUnificationServerHandler(delegatingHttpRequestHandler, detectSsl, false)); } else { - pipeline.addLast("decoder", new HttpRequestDecoder()); - pipeline.addLast("aggregator", new HttpChunkAggregator(1048576)); - pipeline.addLast("encoder", new HttpResponseEncoder()); - pipeline.addLast("deflater", new HttpContentCompressor()); + NettyUtil.initHttpHandlers(pipeline); pipeline.addLast("handler", delegatingHttpRequestHandler); } } // must be after new channels handlers addition (netty bug?) pipeline.remove(this); - Channels.fireMessageReceived(context, buffer, remoteAddress); + context.fireChannelRead(buffer); } @Override - public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) throws Exception { - NettyUtil.log(event.getCause(), BuiltInServer.LOG); + public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception { + NettyUtil.log(cause, BuiltInServer.LOG); } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/Responses.java b/platform/platform-impl/src/org/jetbrains/io/Responses.java index 55f603a22c40..dfb61227e9cf 100644 --- a/platform/platform-impl/src/org/jetbrains/io/Responses.java +++ b/platform/platform-impl/src/org/jetbrains/io/Responses.java @@ -19,14 +19,14 @@ import com.intellij.openapi.application.Application; import com.intellij.openapi.application.ApplicationManager; import com.intellij.openapi.application.ex.ApplicationInfoEx; import com.intellij.openapi.util.text.StringUtil; -import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.buffer.ChannelBuffers; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.ChannelFuture; -import org.jboss.netty.channel.ChannelFutureListener; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.handler.codec.http.*; -import org.jboss.netty.util.CharsetUtil; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.*; +import io.netty.util.CharsetUtil; import org.jetbrains.annotations.Nullable; import java.nio.charset.Charset; @@ -37,11 +37,7 @@ import java.util.Date; import java.util.Locale; import java.util.TimeZone; -import static org.jboss.netty.handler.codec.http.HttpHeaders.Names.*; -import static org.jboss.netty.handler.codec.http.HttpHeaders.isKeepAlive; -import static org.jboss.netty.handler.codec.http.HttpHeaders.setContentLength; -import static org.jboss.netty.handler.codec.http.HttpResponseStatus.OK; -import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1; +import static io.netty.handler.codec.http.HttpHeaders.Names.*; public final class Responses { static final ThreadLocal DATE_FORMAT = new ThreadLocal() { @@ -57,7 +53,7 @@ public final class Responses { private static String SERVER_HEADER_VALUE; public static void addAllowAnyOrigin(HttpResponse response) { - response.setHeader(ACCESS_CONTROL_ALLOW_ORIGIN, "*"); + response.headers().add(ACCESS_CONTROL_ALLOW_ORIGIN, "*"); } public static void addDate(HttpResponse response) { @@ -65,7 +61,12 @@ public final class Responses { } public static void addDate(HttpResponse response, Date date) { - response.setHeader(DATE, DATE_FORMAT.get().format(date)); + response.headers().add(DATE, DATE_FORMAT.get().format(date)); + } + + public static void addNoCache(HttpResponse response) { + response.headers().add(CACHE_CONTROL, "no-cache, no-store, must-revalidate, max-age=0"); + response.headers().add(PRAGMA, "no-cache"); } @Nullable @@ -81,32 +82,29 @@ public final class Responses { public static void addServer(HttpResponse response) { if (getServerHeaderValue() != null) { - response.setHeader(SERVER, getServerHeaderValue()); + response.headers().add(SERVER, getServerHeaderValue()); } } - public static void send(String contentType, CharSequence content, HttpRequest request, ChannelHandlerContext context) { - HttpResponse response = create(contentType); - response.setContent(ChannelBuffers.copiedBuffer(content, CharsetUtil.UTF_8)); - send(response, request, context); - } - - public static void send(HttpResponse response, @Nullable HttpRequest request, ChannelHandlerContext context) { - send(response, context.getChannel(), request); + public static void send(String contentType, CharSequence content, Channel channel, HttpRequest request) { + send(create(contentType, Unpooled.copiedBuffer(content, CharsetUtil.UTF_8)), channel, request); } public static void send(HttpResponse response, Channel channel, @Nullable HttpRequest request) { - ChannelBuffer content = response.getContent(); - setContentLength(response, content == ChannelBuffers.EMPTY_BUFFER ? 0 : content.readableBytes()); + if (response instanceof FullHttpResponse) { + HttpHeaders.setContentLength(response, ((FullHttpResponse)response).content().readableBytes()); + } + else { + HttpHeaders.setContentLength(response, 0); + } - boolean keepAlive = request != null && addKeepAliveIfNeed(response, request); addCommonHeaders(response); - send(response, channel, !keepAlive); + send(response, channel, request != null && !addKeepAliveIfNeed(response, request)); } public static boolean addKeepAliveIfNeed(HttpResponse response, HttpRequest request) { - if (isKeepAlive(request)) { - response.setHeader(CONNECTION, HttpHeaders.Values.KEEP_ALIVE); + if (HttpHeaders.isKeepAlive(request)) { + HttpHeaders.setKeepAlive(response, true); return true; } return false; @@ -119,32 +117,33 @@ public final class Responses { } public static HttpResponse create(String contentType) { - HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); - response.setHeader(CONTENT_TYPE, contentType); + return create(contentType, null); + } + + public static HttpResponse create(@Nullable String contentType, @Nullable ByteBuf content) { + HttpResponse response = content == null + ? new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK) + : new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content); + if (contentType != null) { + response.headers().add(CONTENT_TYPE, contentType); + } return response; } - public static void send(CharSequence content, HttpRequest request, ChannelHandlerContext context) { - send(content, CharsetUtil.US_ASCII, request, context); + public static void send(CharSequence content, Channel channel, @Nullable HttpRequest request) { + send(content, CharsetUtil.US_ASCII, channel, request); } - public static void send(CharSequence content, Charset charset, HttpRequest request, ChannelHandlerContext context) { - DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); - response.setContent(ChannelBuffers.copiedBuffer(content, charset)); - send(response, request, context); + public static void send(CharSequence content, Charset charset, Channel channel, @Nullable HttpRequest request) { + send(create(null, Unpooled.copiedBuffer(content, charset)), channel, request); } - public static void send(byte[] bytes, HttpResponse response, HttpRequest request, ChannelHandlerContext context) { - response.setContent(ChannelBuffers.wrappedBuffer(bytes)); - send(response, request, context); - } - - public static void send(HttpResponse response, ChannelHandlerContext context) { - send(response, context.getChannel(), true); + public static void send(HttpResponse response, Channel channel) { + send(response, channel, true); } public static void send(HttpResponseStatus status, ChannelHandlerContext context) { - send(new DefaultHttpResponse(HTTP_1_1, status), context); + send(new DefaultHttpResponse(HttpVersion.HTTP_1_1, status), context.channel()); } private static void send(HttpResponse response, Channel channel, boolean close) { @@ -152,50 +151,52 @@ public final class Responses { return; } - ChannelFuture future = channel.write(response); + ChannelFuture future = channel.writeAndFlush(response); if (close) { future.addListener(ChannelFutureListener.CLOSE); } } - public static void sendStatus(HttpRequest request, Channel channel, HttpResponseStatus responseStatus, @Nullable String description) { - sendStatus(new DefaultHttpResponse(HTTP_1_1, responseStatus), request, channel, description); + public static void sendStatus(HttpResponseStatus responseStatus, Channel channel) { + sendStatus(null, channel, responseStatus); } - public static void sendStatus(HttpRequest request, Channel channel, HttpResponseStatus responseStatus) { + public static void sendStatus(@Nullable HttpRequest request, Channel channel, HttpResponseStatus responseStatus) { sendStatus(request, channel, responseStatus, null); } + public static void sendStatus(@Nullable HttpRequest request, Channel channel, HttpResponseStatus responseStatus, @Nullable String description) { + send(createStatusResponse(responseStatus, request, description), channel, request); + } + public static void sendStatus(HttpResponse response, HttpRequest request, Channel channel) { - sendStatus(response, request, channel, null); - } - - public static void sendStatus(HttpResponseStatus responseStatus, Channel channel) { - sendStatus(new DefaultHttpResponse(HTTP_1_1, responseStatus), null, channel, null); - } - - private static void sendStatus(HttpResponse response, @Nullable HttpRequest request, Channel channel, @Nullable String description) { - response.setHeader(CONTENT_TYPE, "text/html"); - if (request == null || request.getMethod() != HttpMethod.HEAD) { - String message = response.getStatus().toString(); - - StringBuilder builder = new StringBuilder(); - builder.append("").append(message).append("").append("

").append(message).append("

"); - if (description != null) { - builder.append("

").append(description).append("

"); - } - builder.append("

").append(StringUtil.notNullize(getServerHeaderValue(), "")).append("

"); - - response.setContent(ChannelBuffers.copiedBuffer(builder, CharsetUtil.UTF_8)); - } + response.headers().set(CONTENT_TYPE, "text/html"); send(response, channel, request); } + private static HttpResponse createStatusResponse(HttpResponseStatus responseStatus, @Nullable HttpRequest request, @Nullable String description) { + if (request != null && request.getMethod() == HttpMethod.HEAD) { + return new DefaultHttpResponse(HttpVersion.HTTP_1_1, responseStatus); + } + + StringBuilder builder = new StringBuilder(); + String message = responseStatus.toString(); + builder.append("").append(message).append("").append("

").append(message).append("

"); + if (description != null) { + builder.append("

").append(description).append("

"); + } + builder.append("

").append(StringUtil.notNullize(getServerHeaderValue(), "")).append("

"); + + DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, responseStatus, Unpooled.copiedBuffer(builder, CharsetUtil.UTF_8)); + response.headers().set(CONTENT_TYPE, "text/html"); + return response; + } + public static void sendOptionsResponse(String allowHeaders, HttpRequest request, ChannelHandlerContext context) { - HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); - response.setHeader(ACCESS_CONTROL_ALLOW_ORIGIN, "*"); - response.setHeader(ACCESS_CONTROL_ALLOW_METHODS, allowHeaders); - response.setHeader(ALLOW, allowHeaders); - send(response, request, context); + HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); + addAllowAnyOrigin(response); + response.headers().set(ACCESS_CONTROL_ALLOW_METHODS, allowHeaders); + response.headers().set(ALLOW, allowHeaders); + send(response, context.channel(), request); } } \ No newline at end of file diff --git a/platform/platform-impl/src/org/jetbrains/io/SubServer.java b/platform/platform-impl/src/org/jetbrains/io/SubServer.java index 12512be2efd4..ee070e75f453 100644 --- a/platform/platform-impl/src/org/jetbrains/io/SubServer.java +++ b/platform/platform-impl/src/org/jetbrains/io/SubServer.java @@ -17,24 +17,23 @@ package org.jetbrains.io; import com.intellij.openapi.Disposable; import com.intellij.util.net.NetUtils; -import org.jboss.netty.bootstrap.ServerBootstrap; -import org.jboss.netty.channel.group.ChannelGroup; -import org.jboss.netty.channel.group.DefaultChannelGroup; -import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.EventLoopGroup; import org.jetbrains.ide.BuiltInServerManager; import org.jetbrains.ide.CustomPortServerManager; import java.net.InetSocketAddress; final class SubServer implements CustomPortServerManager.CustomPortService, Disposable { - private final ChannelGroup openChannels = new DefaultChannelGroup(); + private final ChannelRegistrar channelRegistrar = new ChannelRegistrar(); + private final CustomPortServerManager user; private final ServerBootstrap bootstrap; - public SubServer(CustomPortServerManager user, NioServerSocketChannelFactory channelFactory) { + public SubServer(CustomPortServerManager user, EventLoopGroup eventLoopGroup) { this.user = user; user.setManager(this); - bootstrap = BuiltInServer.createServerBootstrap(channelFactory, openChannels, user.createXmlRpcHandlers()); + bootstrap = BuiltInServer.createServerBootstrap(eventLoopGroup, channelRegistrar, user.createXmlRpcHandlers()); } public boolean bind(int port) { @@ -43,7 +42,8 @@ final class SubServer implements CustomPortServerManager.CustomPortService, Disp } try { - openChannels.add(bootstrap.bind(user.isAvailableExternally() ? new InetSocketAddress(port) : new InetSocketAddress(NetUtils.getLoopbackAddress(), port))); + bootstrap.localAddress(user.isAvailableExternally() ? new InetSocketAddress(port) : new InetSocketAddress(NetUtils.getLoopbackAddress(), port)); + channelRegistrar.add(bootstrap.bind().syncUninterruptibly().channel()); return true; } catch (Exception e) { @@ -55,12 +55,11 @@ final class SubServer implements CustomPortServerManager.CustomPortService, Disp @Override public boolean isBound() { - return !openChannels.isEmpty(); + return !channelRegistrar.isEmpty(); } private void stop() { - openChannels.close().awaitUninterruptibly(); - openChannels.clear(); + channelRegistrar.close(false); } @Override diff --git a/platform/remote-servers/impl/src/com/intellij/remoteServer/impl/runtime/ServerTaskExecutorImpl.java b/platform/remote-servers/impl/src/com/intellij/remoteServer/impl/runtime/ServerTaskExecutorImpl.java index 43722962fcbd..d9c717fd4125 100644 --- a/platform/remote-servers/impl/src/com/intellij/remoteServer/impl/runtime/ServerTaskExecutorImpl.java +++ b/platform/remote-servers/impl/src/com/intellij/remoteServer/impl/runtime/ServerTaskExecutorImpl.java @@ -13,11 +13,10 @@ import org.jetbrains.ide.PooledThreadExecutor; */ public class ServerTaskExecutorImpl implements ServerTaskExecutor { private static final Logger LOG = Logger.getInstance(ServerTaskExecutorImpl.class); - private static final PooledThreadExecutor POOLED_THREAD_EXECUTOR = new PooledThreadExecutor(); private final SequentialTaskExecutor myTaskExecutor; public ServerTaskExecutorImpl() { - myTaskExecutor = new SequentialTaskExecutor(POOLED_THREAD_EXECUTOR); + myTaskExecutor = new SequentialTaskExecutor(PooledThreadExecutor.INSTANCE); } @Override