Add websocket support to HTTP API

This uses Netty's websocket functionality, meaning we do not have to
depend on another library.

As websockets do not fit neatly into the standard polling socket model,
the API is significantly more event based than CCTweaks's. One uses
http.websocket to connect, which will wait until a connection is
established and then returns the connection object (an async variant is
available).

Once you have a websocket object, you can use .send(msg) to transmit a
message. Incoming messages will fire a "websocket_message" event, with
the URL and content as arguments. A convenience method (.receive())
exists to aid waiting for valid messages.

src/main/java/dan200/computercraft/ComputerCraft.java

@@ -121,6 +121,7 @@ public class ComputerCraft
     };
     
     public static boolean http_enable = true;
+    public static boolean http_websocket_enable = true;
     public static AddressPredicate http_whitelist = new AddressPredicate( DEFAULT_HTTP_WHITELIST );
     public static AddressPredicate http_blacklist = new AddressPredicate( DEFAULT_HTTP_BLACKLIST );
     public static boolean disable_lua51_features = false;
@@ -200,6 +201,7 @@ public class ComputerCraft
         public static Configuration config;
 
         public static Property http_enable;
+        public static Property http_websocket_enable;
         public static Property http_whitelist;
         public static Property http_blacklist;
         public static Property disable_lua51_features;
@@ -271,6 +273,9 @@ public class ComputerCraft
         Config.http_enable = Config.config.get( Configuration.CATEGORY_GENERAL, "http_enable", http_enable );
         Config.http_enable.setComment( "Enable the \"http\" API on Computers (see \"http_whitelist\" and \"http_blacklist\" for more fine grained control than this)" );
 
+        Config.http_websocket_enable = Config.config.get( Configuration.CATEGORY_GENERAL, "http_websocket_enable", http_websocket_enable );
+        Config.http_websocket_enable.setComment( "Enable use of http websockets. This requires the \"http_enable\" option to also be true." );
+
         {
             ConfigCategory category = Config.config.getCategory( Configuration.CATEGORY_GENERAL );
             Property currentProperty = category.get( "http_whitelist" );
@@ -362,6 +367,7 @@ public class ComputerCraft
     public static void syncConfig() {
 
         http_enable = Config.http_enable.getBoolean();
+        http_websocket_enable = Config.http_websocket_enable.getBoolean();
         http_whitelist = new AddressPredicate( Config.http_whitelist.getStringList() );
         http_blacklist = new AddressPredicate( Config.http_blacklist.getStringList() );
         disable_lua51_features = Config.disable_lua51_features.getBoolean();

src/main/java/dan200/computercraft/core/apis/HTTPAPI.java

@@ -6,13 +6,18 @@
 
 package dan200.computercraft.core.apis;
 
+import dan200.computercraft.ComputerCraft;
 import dan200.computercraft.api.lua.ILuaContext;
 import dan200.computercraft.api.lua.LuaException;
 import dan200.computercraft.core.apis.http.HTTPCheck;
-import dan200.computercraft.core.apis.http.HTTPRequest;
 import dan200.computercraft.core.apis.http.HTTPExecutor;
+import dan200.computercraft.core.apis.http.HTTPRequest;
+import dan200.computercraft.core.apis.http.WebsocketConnector;
 
 import javax.annotation.Nonnull;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.URI;
 import java.net.URL;
 import java.util.*;
 import java.util.concurrent.Future;
@@ -23,13 +28,15 @@ public class HTTPAPI implements ILuaAPI
 {
     private final IAPIEnvironment m_apiEnvironment;
     private final List<Future<?>> m_httpTasks;
-    
+    private final Set<Closeable> m_closeables;
+
     public HTTPAPI( IAPIEnvironment environment )
     {
         m_apiEnvironment = environment;
         m_httpTasks = new ArrayList<>();
+        m_closeables = new HashSet<>();
     }
-    
+
     @Override
     public String[] getNames()
     {
@@ -69,15 +76,30 @@ public class HTTPAPI implements ILuaAPI
             }
             m_httpTasks.clear();
         }
+        synchronized( m_closeables )
+        {
+            for( Closeable x : m_closeables )
+            {
+                try
+                {
+                    x.close();
+                }
+                catch( IOException ignored )
+                {
+                }
+            }
+            m_closeables.clear();
+        }
     }
 
     @Nonnull
     @Override
     public String[] getMethodNames()
     {
-         return new String[] {
+        return new String[] {
             "request",
-            "checkURL"
+            "checkURL",
+            "websocket",
         };
     }
 
@@ -156,10 +178,63 @@ public class HTTPAPI implements ILuaAPI
                     return new Object[] { false, e.getMessage() };
                 }
             }
+            case 2: // websocket
+            {
+                String address = getString( args, 0 );
+                Map<Object, Object> headerTbl = optTable( args, 1, Collections.emptyMap() );
+
+                HashMap<String, String> headers = new HashMap<String, String>( headerTbl.size() );
+                for( Object key : headerTbl.keySet() )
+                {
+                    Object value = headerTbl.get( key );
+                    if( key instanceof String && value instanceof String )
+                    {
+                        headers.put( (String) key, (String) value );
+                    }
+                }
+
+                if( !ComputerCraft.http_websocket_enable )
+                {
+                    throw new LuaException( "Websocket connections are disabled" );
+                }
+
+                try
+                {
+                    URI uri = WebsocketConnector.checkURI( address );
+                    int port = WebsocketConnector.getPort( uri );
+
+                    Future<?> connector = WebsocketConnector.createConnector( m_apiEnvironment, this, uri, address, port, headers );
+                    synchronized( m_httpTasks )
+                    {
+                        m_httpTasks.add( connector );
+                    }
+                    return new Object[] { true };
+                }
+                catch( HTTPRequestException e )
+                {
+                    return new Object[] { false, e.getMessage() };
+                }
+            }
             default:
             {
                 return null;
             }
         }
     }
+
+    public void addCloseable( Closeable closeable )
+    {
+        synchronized( m_closeables )
+        {
+            m_closeables.add( closeable );
+        }
+    }
+
+    public void removeCloseable( Closeable closeable )
+    {
+        synchronized( m_closeables )
+        {
+            m_closeables.remove( closeable );
+        }
+    }
 }

src/main/java/dan200/computercraft/core/apis/http/HTTPExecutor.java

@@ -9,6 +9,8 @@ package dan200.computercraft.core.apis.http;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
 
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -30,6 +32,13 @@ public final class HTTPExecutor
             .build()
     ) );
 
+    public static final EventLoopGroup LOOP_GROUP = new NioEventLoopGroup( 4, new ThreadFactoryBuilder()
+        .setDaemon( true )
+        .setPriority( Thread.MIN_PRIORITY + (Thread.NORM_PRIORITY - Thread.MIN_PRIORITY) / 2 )
+        .setNameFormat( "ComputerCraft-Netty-%d" )
+        .build()
+    );
+
     private HTTPExecutor()
     {
     }

src/main/java/dan200/computercraft/core/apis/http/WebsocketConnection.java

@@ -0,0 +1,186 @@
+/*
+ * This file is part of ComputerCraft - http://www.computercraft.info
+ * Copyright Daniel Ratcliffe, 2011-2017. Do not distribute without permission.
+ * Send enquiries to dratcliffe@gmail.com
+ */
+
+package dan200.computercraft.core.apis.http;
+
+import com.google.common.base.Objects;
+import dan200.computercraft.api.lua.ILuaContext;
+import dan200.computercraft.api.lua.ILuaObject;
+import dan200.computercraft.api.lua.LuaException;
+import dan200.computercraft.core.apis.HTTPAPI;
+import dan200.computercraft.core.apis.IAPIEnvironment;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.websocketx.*;
+import io.netty.util.CharsetUtil;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.IOException;
+
+public class WebsocketConnection extends SimpleChannelInboundHandler<Object> implements ILuaObject, Closeable
+{
+    public static final String SUCCESS_EVENT = "websocket_success";
+    public static final String FAILURE_EVENT = "websocket_failure";
+    public static final String CLOSE_EVENT = "websocket_closed";
+    public static final String MESSAGE_EVENT = "websocket_message";
+
+    private final String url;
+    private final IAPIEnvironment computer;
+    private final HTTPAPI api;
+    private boolean open = true;
+
+    private Channel channel;
+    private final WebSocketClientHandshaker handshaker;
+
+    public WebsocketConnection( IAPIEnvironment computer, HTTPAPI api, WebSocketClientHandshaker handshaker, String url )
+    {
+        this.computer = computer;
+        this.api = api;
+        this.handshaker = handshaker;
+        this.url = url;
+
+        api.addCloseable( this );
+    }
+
+    private void close( boolean remove )
+    {
+        open = false;
+        if( remove ) api.removeCloseable( this );
+
+        if( channel != null )
+        {
+            channel.close();
+            channel = null;
+        }
+    }
+
+    @Override
+    public void close() throws IOException
+    {
+        close( false );
+    }
+
+    private void onClosed()
+    {
+        close( true );
+        computer.queueEvent( CLOSE_EVENT, new Object[] { url } );
+    }
+
+    @Override
+    public void handlerAdded( ChannelHandlerContext ctx ) throws Exception
+    {
+        channel = ctx.channel();
+        super.handlerAdded( ctx );
+    }
+
+    @Override
+    public void channelActive( ChannelHandlerContext ctx ) throws Exception
+    {
+        handshaker.handshake( ctx.channel() );
+        super.channelActive( ctx );
+    }
+
+    @Override
+    public void channelInactive( ChannelHandlerContext ctx ) throws Exception
+    {
+        onClosed();
+        super.channelInactive( ctx );
+    }
+
+    @Override
+    public void channelRead0( ChannelHandlerContext ctx, Object msg ) throws Exception
+    {
+        Channel ch = ctx.channel();
+        if( !handshaker.isHandshakeComplete() )
+        {
+            handshaker.finishHandshake( ch, (FullHttpResponse) msg );
+            computer.queueEvent( SUCCESS_EVENT, new Object[] { url, this } );
+            return;
+        }
+
+        if( msg instanceof FullHttpResponse )
+        {
+            FullHttpResponse response = (FullHttpResponse) msg;
+            throw new IllegalStateException( "Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString( CharsetUtil.UTF_8 ) + ')' );
+        }
+
+        WebSocketFrame frame = (WebSocketFrame) msg;
+        if( frame instanceof TextWebSocketFrame )
+        {
+            computer.queueEvent( MESSAGE_EVENT, new Object[] { url, ((TextWebSocketFrame) frame).text() } );
+        }
+        else if( frame instanceof BinaryWebSocketFrame )
+        {
+            ByteBuf data = frame.content();
+            byte[] converted = new byte[ data.readableBytes() ];
+            data.readBytes( converted );
+            computer.queueEvent( MESSAGE_EVENT, new Object[] { url, data } );
+        }
+        else if( frame instanceof CloseWebSocketFrame )
+        {
+            ch.close();
+            onClosed();
+        }
+    }
+
+    @Override
+    public void exceptionCaught( ChannelHandlerContext ctx, Throwable cause )
+    {
+        ctx.close();
+        computer.queueEvent( FAILURE_EVENT, new Object[] {
+            url,
+            cause instanceof WebSocketHandshakeException ? cause.getMessage() : "Could not connect"
+        } );
+    }
+
+    @Nonnull
+    @Override
+    public String[] getMethodNames()
+    {
+        return new String[] { "receive", "send", "close" };
+    }
+
+    @Nullable
+    @Override
+    public Object[] callMethod( @Nonnull ILuaContext context, int method, @Nonnull Object[] arguments ) throws LuaException, InterruptedException
+    {
+        switch( method )
+        {
+            case 0:
+                while( true )
+                {
+                    checkOpen();
+                    Object[] event = context.pullEvent( MESSAGE_EVENT );
+                    if( event.length >= 3 && Objects.equal( event[ 1 ], url ) )
+                    {
+                        return new Object[] { event[ 2 ] };
+                    }
+                }
+            case 1:
+            {
+                checkOpen();
+                String text = arguments.length > 0 && arguments[ 0 ] != null ? arguments[ 0 ].toString() : "";
+                channel.writeAndFlush( new TextWebSocketFrame( text ) );
+                return null;
+            }
+            case 2:
+                close( true );
+                return null;
+            default:
+                return null;
+        }
+    }
+
+    private void checkOpen() throws LuaException
+    {
+        if( !open ) throw new LuaException( "attempt to use a closed file" );
+    }
+}

src/main/java/dan200/computercraft/core/apis/http/WebsocketConnector.java

@@ -0,0 +1,201 @@
+/*
+ * This file is part of ComputerCraft - http://www.computercraft.info
+ * Copyright Daniel Ratcliffe, 2011-2017. Do not distribute without permission.
+ * Send enquiries to dratcliffe@gmail.com
+ */
+
+package dan200.computercraft.core.apis.http;
+
+import dan200.computercraft.ComputerCraft;
+import dan200.computercraft.core.apis.HTTPAPI;
+import dan200.computercraft.core.apis.HTTPRequestException;
+import dan200.computercraft.core.apis.IAPIEnvironment;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.http.DefaultHttpHeaders;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
+import io.netty.handler.codec.http.websocketx.WebSocketVersion;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslContextBuilder;
+
+import javax.net.ssl.SSLException;
+import javax.net.ssl.TrustManagerFactory;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.security.KeyStore;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+/*
+ * Provides functionality to verify and connect to a remote websocket.
+ */
+public final class WebsocketConnector
+{
+    private static final Object lock = new Object();
+    private static TrustManagerFactory trustManager;
+
+    private WebsocketConnector()
+    {
+    }
+
+    private static TrustManagerFactory getTrustManager()
+    {
+        if( trustManager != null ) return trustManager;
+        synchronized( lock )
+        {
+            if( trustManager != null ) return trustManager;
+
+            TrustManagerFactory tmf = null;
+            try
+            {
+                tmf = TrustManagerFactory.getInstance( TrustManagerFactory.getDefaultAlgorithm() );
+                tmf.init( (KeyStore) null );
+            }
+            catch( Exception e )
+            {
+                ComputerCraft.log.error( "Cannot setup trust manager", e );
+            }
+
+            return trustManager = tmf;
+        }
+    }
+
+    public static URI checkURI( String address ) throws HTTPRequestException
+    {
+        URI uri = null;
+        try
+        {
+            uri = new URI( address );
+        }
+        catch( URISyntaxException ignored )
+        {
+        }
+
+        if( uri == null || uri.getHost() == null )
+        {
+            try
+            {
+                uri = new URI( "ws://" + address );
+            }
+            catch( URISyntaxException ignored )
+            {
+            }
+        }
+
+        if( uri == null || uri.getHost() == null ) throw new HTTPRequestException( "URL malformed" );
+
+        String scheme = uri.getScheme();
+        if( scheme == null )
+        {
+            try
+            {
+                uri = new URI( "ws://" + uri.toString() );
+            }
+            catch( URISyntaxException e )
+            {
+                throw new HTTPRequestException( "URL malformed" );
+            }
+        }
+        else if( !scheme.equalsIgnoreCase( "wss" ) && !scheme.equalsIgnoreCase( "ws" ) )
+        {
+            throw new HTTPRequestException( "Invalid scheme '" + scheme + "'" );
+        }
+
+        if( !ComputerCraft.http_whitelist.matches( uri.getHost() ) || ComputerCraft.http_blacklist.matches( uri.getHost() ) )
+        {
+            throw new HTTPRequestException( "Domain not permitted" );
+        }
+
+        return uri;
+    }
+
+    public static int getPort( URI uri ) throws HTTPRequestException
+    {
+        int port = uri.getPort();
+        if( port >= 0 ) return port;
+
+        String scheme = uri.getScheme();
+        if( scheme.equalsIgnoreCase( "ws" ) )
+        {
+            return 80;
+        }
+        else if( scheme.equalsIgnoreCase( "wss" ) )
+        {
+            return 443;
+        }
+        else
+        {
+            throw new HTTPRequestException( "Invalid scheme '" + scheme + "'" );
+        }
+    }
+
+    public static Future<?> createConnector( final IAPIEnvironment environment, final HTTPAPI api, final URI uri, final String address, final int port, final Map<String, String> headers )
+    {
+        return HTTPExecutor.EXECUTOR.submit( () -> {
+            InetAddress resolved;
+            try
+            {
+                resolved = HTTPRequest.checkHost( uri.getHost() );
+            }
+            catch( HTTPRequestException e )
+            {
+                environment.queueEvent( WebsocketConnection.FAILURE_EVENT, new Object[] { address, e.getMessage() } );
+                return;
+            }
+
+            InetSocketAddress socketAddress = new InetSocketAddress( resolved, uri.getPort() == -1 ? port : uri.getPort() );
+
+            final SslContext ssl;
+            if( uri.getScheme().equalsIgnoreCase( "wss" ) )
+            {
+                try
+                {
+                    ssl = SslContextBuilder.forClient().trustManager( getTrustManager() ).build();
+                }
+                catch( SSLException e )
+                {
+                    environment.queueEvent( WebsocketConnection.FAILURE_EVENT, new Object[] { address, "Cannot create secure socket" } );
+                    return;
+                }
+            }
+            else
+            {
+                ssl = null;
+            }
+
+            HttpHeaders httpHeaders = new DefaultHttpHeaders();
+            for( Map.Entry<String, String> header : headers.entrySet() )
+            {
+                httpHeaders.add( header.getKey(), header.getValue() );
+            }
+
+            WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker( uri, WebSocketVersion.V13, null, false, httpHeaders );
+            final WebsocketConnection connection = new WebsocketConnection( environment, api, handshaker, address );
+
+            new Bootstrap()
+                .group( HTTPExecutor.LOOP_GROUP )
+                .channel( NioSocketChannel.class )
+                .handler( new ChannelInitializer<SocketChannel>()
+                {
+                    @Override
+                    protected void initChannel( SocketChannel ch ) throws Exception
+                    {
+                        ChannelPipeline p = ch.pipeline();
+                        if( ssl != null ) p.addLast( ssl.newHandler( ch.alloc(), uri.getHost(), port ) );
+                        p.addLast( new HttpClientCodec(), new HttpObjectAggregator( 8192 ), connection );
+                    }
+                } )
+                .remoteAddress( socketAddress )
+                .connect();
+        } );
+    }
+}

src/main/resources/assets/computercraft/lang/en_us.lang

@@ -43,6 +43,7 @@ gui.computercraft:wired_modem.peripheral_connected=Peripheral "%s" connected to
 gui.computercraft:wired_modem.peripheral_disconnected=Peripheral "%s" disconnected from network
 
 gui.computercraft:config.http_enable=Enable HTTP API
+gui.computercraft:config.http_websocket_enable=Enable HTTP websockets
 gui.computercraft:config.http_whitelist=HTTP whitelist
 gui.computercraft:config.http_blacklist=HTTP blacklist
 gui.computercraft:config.disable_lua51_features=Disable Lua 5.1 features