MDS/openide
1
0
Fork 0
mirror of https://gitflic.ru/project/openide/openide.git synced 2026-09-27 10:03:11 +07:00
Code Issues Packages Projects Releases Wiki Activity

netty 4

Browse Source
This commit is contained in:
Vladimir Krivosheev
2013-08-20 18:02:36 +02:00
parent 9290bbcc99
commit 95465fad47
40 changed files with 672 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.idea/libraries/Netty.xml
Generated
+2 -2
View File
@@ -1,11 +1,11 @@
<component name="libraryTable">
<library name="Netty">
<CLASSES>
<root url="jar://$PROJECT_DIR$/lib/netty-3.6.6.Final.jar!/" />
<root url="jar://$PROJECT_DIR$/lib/netty-all.jar!/" />
</CLASSES>
<JAVADOC />
<SOURCES>
<root url="jar://$PROJECT_DIR$/lib/src/netty-3.6.6.Final-sources.jar!/" />
<root url="jar://$PROJECT_DIR$/lib/src/netty-all-sources.jar!/" />
</SOURCES>
</library>
</component>
build/scripts/libLicenses.gant
+1 -1
View File
@@ -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")
java/compiler/impl/src/com/intellij/compiler/impl/ProblemsViewImpl.java
+1 -1
View File
@@ -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);
java/compiler/impl/src/com/intellij/compiler/server/BuildManager.java
+31 -52
View File
@@ -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<Boolean> ALLOW_AUTOMAKE = Key.create("_allow_automake_when_process_is_active_");
private static final Key<String> FORCE_MODEL_LOADING_PARAMETER = Key.create(BuildParametersKeys.FORCE_MODEL_LOADING);
@@ -152,8 +152,7 @@ public class BuildManager implements ApplicationComponent{
private final Map<RequestFuture, Project> myAutomakeFutures = new HashMap<RequestFuture, Project>();
private final Map<String, RequestFuture> myBuildsInProgress = Collections.synchronizedMap(new HashMap<String, RequestFuture>());
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<String, ProjectData> myProjectDataMap = Collections.synchronizedMap(new HashMap<String, ProjectData>());
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
java/compiler/impl/src/com/intellij/compiler/server/BuildMessageDispatcher.java
+23 -39
View File
@@ -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<CmdlineRemoteProto.Message> {
private static final Logger LOG = Logger.getInstance("#com.intellij.compiler.server.BuildMessageDispatcher");
private static final AttributeKey<SessionData> SESSION_DATA = new AttributeKey<SessionData>("BuildMessageDispatcher.sessionData");
private final Map<UUID, SessionData> myMessageHandlers = new ConcurrentHashMap<UUID, SessionData>();
private final Set<UUID> myCanceledSessions = new ConcurrentHashSet<UUID>();
@@ -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 {
java/compiler/impl/src/com/intellij/compiler/server/BuilderMessageHandler.java
+1 -1
View File
@@ -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;
java/compiler/impl/src/com/intellij/compiler/server/DefaultMessageHandler.java
+12 -11
View File
@@ -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<PsiIdentifier> 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();
}
java/compiler/impl/src/com/intellij/compiler/server/MessageHandlerWrapper.java
+1 -1
View File
@@ -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;
jps/jps-builders/src/org/jetbrains/jps/client/ProtobufClientMessageHandler.java
+28 -29
View File
@@ -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<T extends ProtobufResponseHandler> extends SimpleChannelHandler {
@ChannelHandler.Sharable
final class ProtobufClientMessageHandler<T extends ProtobufResponseHandler> extends SimpleChannelInboundHandler<MessageLite> {
private final ConcurrentHashMap<UUID, RequestFuture<T>> myHandlers = new ConcurrentHashMap<UUID, RequestFuture<T>>();
@NotNull
private final UUIDGetter myUuidGetter;
@@ -45,8 +46,9 @@ final class ProtobufClientMessageHandler<T extends ProtobufResponseHandler> 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<T> future = myHandlers.get(messageUUID);
final T handler = future != null ? future.getMessageHandler() : null;
if (handler == null) {
@@ -55,7 +57,7 @@ final class ProtobufClientMessageHandler<T extends ProtobufResponseHandler> 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<T extends ProtobufResponseHandler> exte
try {
handler.sessionTerminated();
}
catch (Throwable ignored) {
ignored.printStackTrace();
catch (Throwable e) {
//noinspection CallToPrintStackTrace
e.printStackTrace();
}
}
}
@@ -89,30 +92,26 @@ final class ProtobufClientMessageHandler<T extends ProtobufResponseHandler> exte
}
}
public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
try {
super.channelClosed(ctx, e);
}
finally {
for (UUID uuid : new ArrayList<UUID>(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<UUID>(myHandlers.keySet())) {
terminateSession(uuid);
}
});
}
finally {
// make sure the client is in disconnected state
myAsyncExec.execute(new Runnable() {
@Override
public void run() {
myClient.disconnect();
}
});
}
}
}
jps/jps-builders/src/org/jetbrains/jps/client/SimpleProtobufClient.java
+28 -40
View File
@@ -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<T extends ProtobufResponseHandler> {
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<State> myState = new AtomicReference<State>(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<T> myMessageHandler;
public SimpleProtobufClient(final MessageLite msgDefaultInstance, final Executor asyncExec, final UUIDGetter uuidGetter) {
myMessageHandler = new ProtobufClientMessageHandler<T>(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<T extends ProtobufResponseHandler> {
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<T extends ProtobufResponseHandler> {
LOG.error(e);
}
}
else {
final Throwable reason = future.getCause();
if (reason != null) {
throw reason;
}
}
return success;
}
finally {
@@ -129,7 +116,7 @@ public class SimpleProtobufClient<T extends ProtobufResponseHandler> {
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<T extends ProtobufResponseHandler> {
final RequestFuture<T> requestFuture = new RequestFuture<T>(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);
jps/jps-builders/src/org/jetbrains/jps/client/UUIDGetter.java
+2 -2
View File
@@ -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);
}
jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildMain.java
+39 -46
View File
@@ -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<CmdlineRemoteProto.Message> {
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 {
jps/jps-builders/src/org/jetbrains/jps/cmdline/BuildSession.java
+10 -11
View File
@@ -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<Boolean> hasErrors = new Ref<Boolean>(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;
}
}
jps/jps-builders/src/org/jetbrains/jps/cmdline/ClasspathBootstrap.java
+5 -5
View File
@@ -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<StandardJavaFileManager> c = (Class<StandardJavaFileManager>)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<StandardJavaFileManager> c = (Class<StandardJavaFileManager>)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<StandardJavaFileManager> optimizedFileManagerClass = getOptimizedFileManagerClass();
if (optimizedFileManagerClass != null) {
jps/jps-builders/src/org/jetbrains/jps/javac/JavacServer.java
+95 -71
View File
@@ -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<String> options,
Collection<File> 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<? extends JavaFileObject> 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<String> imports, Collection<String> 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<JavacRemoteProto.Message> {
@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<String> 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;
}
jps/jps-builders/src/org/jetbrains/jps/javac/JavacServerClient.java
+3 -3
View File
@@ -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<JavacServerResponseH
public JavacServerClient() {
super(JavacRemoteProto.Message.getDefaultInstance(), SharedThreadPool.getInstance(), new UUIDGetter() {
@Override
@NotNull
public UUID getSessionUUID(@NotNull MessageEvent e) {
final JavacRemoteProto.Message message = (JavacRemoteProto.Message)e.getMessage();
public UUID getSessionUUID(@NotNull JavacRemoteProto.Message message) {
final JavacRemoteProto.Message.UUID uuid = message.getSessionId();
return new UUID(uuid.getMostSigBits(), uuid.getLeastSigBits());
}
@@ -46,6 +45,7 @@ public class JavacServerClient extends SimpleProtobufClient<JavacServerResponseH
final JavacServerResponseHandler rh = new JavacServerResponseHandler(diagnosticSink, outputSink);
final JavacRemoteProto.Message.Request request = JavacProtoUtil.createCompilationRequest(options, files, classpath, platformCp, sourcePath, outs);
return sendRequest(request, rh, new RequestFuture.CancelAction<JavacServerResponseHandler>() {
@Override
public void cancel(RequestFuture<JavacServerResponseHandler> javacServerResponseHandlerRequestFuture) throws Exception {
sendRequest(JavacProtoUtil.createCancelRequest(), null, null);
}
lib/netty-3.6.6.Final.jar
BIN
View File
Binary file not shown.
lib/netty-all-4.0.7.Final.jar
BIN
View File
Binary file not shown.
lib/netty-all.jar
BIN
View File
Binary file not shown.
lib/required_for_dist.txt
+1 -1
View File
@@ -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
lib/src/netty-3.6.6.Final-sources.jar
BIN
View File
Binary file not shown.
lib/src/netty-all-4.0.7.Final-sources.jar
BIN
View File
Binary file not shown.
lib/src/netty-all-sources.jar
BIN
View File
Binary file not shown.
platform/platform-api/src/com/intellij/ide/XmlRpcServer.java
+3 -3
View File
@@ -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<String, Object> handlers) throws IOException;
boolean process(@NotNull String path, @NotNull FullHttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map<String, Object> handlers) throws IOException;
final class SERVICE {
private SERVICE() {
platform/platform-api/src/org/jetbrains/ide/HttpRequestHandler.java
+6 -6
View File
@@ -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;
}
platform/platform-impl/src/com/intellij/ide/XmlRpcServerImpl.java
+17 -20
View File
@@ -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<String, Object> handlers) throws IOException {
public boolean process(@NotNull String path, @NotNull FullHttpRequest request, @NotNull ChannelHandlerContext context, @Nullable Map<String, Object> 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;
platform/platform-impl/src/org/jetbrains/ide/BuiltInServerManagerImpl.java
+10 -10
View File
@@ -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);
}
}
});
}
platform/platform-impl/src/org/jetbrains/ide/PooledThreadExecutor.java
+6 -1
View File
@@ -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);
platform/platform-impl/src/org/jetbrains/io/BuiltInServer.java
+47 -73
View File
@@ -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<String, Object> 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<String, Object> 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<String, Object> 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);
}
}
}
platform/platform-impl/src/org/jetbrains/io/ChannelExceptionHandler.java
+21 -12
View File
@@ -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())) {
platform/platform-impl/src/org/jetbrains/io/ChannelRegistrar.java
+54
View File
@@ -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();
}
}
}
}
platform/platform-impl/src/org/jetbrains/io/Decoder.java
+14 -10
View File
@@ -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<I> extends SimpleChannelInboundHandler<I> {
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;
platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandler.java
+26 -12
View File
@@ -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<HttpRequestHandler> PREV_HANDLER = new AttributeKey<HttpRequestHandler>("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<HttpRequestHandler> 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();
}
}
platform/platform-impl/src/org/jetbrains/io/DelegatingHttpRequestHandlerBase.java
+13 -23
View File
@@ -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<FullHttpRequest> {
@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();
}
}
}
platform/platform-impl/src/org/jetbrains/io/FileResponses.java
+22 -34
View File
@@ -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 {
}
}
}
}
}
platform/platform-impl/src/org/jetbrains/io/NettyUtil.java
+32 -10
View File
@@ -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());
}
}
platform/platform-impl/src/org/jetbrains/io/PortUnificationServerHandler.java
+30 -50
View File
@@ -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<ByteBuf> {
private static final AtomicNotNullLazyValue<SSLContext> SSL_SERVER_CONTEXT = new AtomicNotNullLazyValue<SSLContext>() {
@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);
}
}
platform/platform-impl/src/org/jetbrains/io/Responses.java
+77 -76
View File
@@ -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<DateFormat> DATE_FORMAT = new ThreadLocal<DateFormat>() {
@@ -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("<!doctype html><title>").append(message).append("</title>").append("<h1 style=\"text-align: center\">").append(message).append("</h1>");
if (description != null) {
builder.append("<p>").append(description).append("</p>");
}
builder.append("<hr/><p style=\"text-align: center\">").append(StringUtil.notNullize(getServerHeaderValue(), "")).append("</p>");
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("<!doctype html><title>").append(message).append("</title>").append("<h1 style=\"text-align: center\">").append(message).append("</h1>");
if (description != null) {
builder.append("<p>").append(description).append("</p>");
}
builder.append("<hr/><p style=\"text-align: center\">").append(StringUtil.notNullize(getServerHeaderValue(), "")).append("</p>");
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);
}
}
platform/platform-impl/src/org/jetbrains/io/SubServer.java
+10 -11
View File
@@ -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
platform/remote-servers/impl/src/com/intellij/remoteServer/impl/runtime/ServerTaskExecutorImpl.java
+1 -2
View File
@@ -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