使用Netty和阿里云函数写一个低成本的“梯子”

package cn.wycode.lambda.proxy.inbound

import cn.wycode.lambda.proxy.config.AliyunConfig
import cn.wycode.lambda.proxy.outbound.ProxyOutboundInitializer
import io.netty.bootstrap.Bootstrap
import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled
import io.netty.channel.*
import io.netty.handler.codec.http.*
import io.netty.util.CharsetUtil

class ProxyInboundHandler(val aliyunConfig: AliyunConfig) : SimpleChannelInboundHandler<ByteBuf>() {


    // As we use inboundChannel.eventLoop() when building the Bootstrap this does not need to be volatile as
    // the outboundChannel will use the same EventLoop (and therefore Thread) as the inboundChannel.
    private var outboundChannel: Channel? = null

    override fun channelRead0(ctx: ChannelHandlerContext, msg: ByteBuf) {
        println("ProxyInboundHandler<<<" + msg.toString(CharsetUtil.UTF_8))
//        val headers = msg.headers()
//        val headerMap = HashMap<String, String>(headers.size())
//        headers.forEach { headerMap[it.key] = it.value }
//        val outboundBody = Request(msg.method().name(), msg.uri(), msg.protocolVersion().text(), headerMap)
//        val outboundJson = JSON.toJSONString(outboundBody)
        val request = DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, aliyunConfig.path)
        request.headers().set(HttpHeaderNames.HOST, aliyunConfig.host)
        request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
        request.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
        request.headers().set(HttpHeaderNames.CONTENT_LENGTH, msg.writerIndex())
//        request.content().writeCharSequence(outboundJson, CharsetUtil.UTF_8)
        request.content().writeBytes(msg)
//
        if (outboundChannel!!.isActive) {
            println("ProxyInboundHandler>>>${request.content().toString(CharsetUtil.UTF_8)}")
            outboundChannel!!.writeAndFlush(request).addListener(object : ChannelFutureListener {
                override fun operationComplete(future: ChannelFuture) {
                    if (future.isSuccess) {
                        // was able to flush out data, start to read the next chunk
                        ctx.channel().read()
                    } else {
                        future.channel().close()
                    }
                }
            })
        }
    }


    override fun channelActive(ctx: ChannelHandlerContext) {
        val inboundChannel = ctx.channel()
        // Start the connection attempt.
        val outboundClient = Bootstrap()
        outboundClient.group(inboundChannel.eventLoop())
                .channel(inboundChannel.javaClass)
                .handler(ProxyOutboundInitializer(inboundChannel))
                .option(ChannelOption.AUTO_READ, false)

        val f = outboundClient.connect(aliyunConfig.host, aliyunConfig.port)
        outboundChannel = f.channel()
        f.addListener { future ->
            if (future.isSuccess) {
                // connection complete start to read first data
                inboundChannel.read()
            } else {
                // Close the connection if the connection attempt has failed.
                inboundChannel.close()
            }
        }
    }


    override fun channelInactive(ctx: ChannelHandlerContext) {
        if (outboundChannel != null) {
            closeOnFlush(outboundChannel!!)
        }
    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
        cause.printStackTrace()
        closeOnFlush(ctx.channel())
    }

    companion object {
        fun closeOnFlush(ch: Channel) {
            if (ch.isActive) {
                ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE)
            }
        }
    }

}

package cn.wycode.lambda.proxy.outbound

import cn.wycode.lambda.proxy.inbound.ProxyInboundHandler
import io.netty.channel.*
import io.netty.handler.codec.http.FullHttpResponse
import io.netty.util.CharsetUtil

class ProxyOutboundHandler(private val inboundChannel: Channel) : SimpleChannelInboundHandler<FullHttpResponse>() {

    override fun channelActive(ctx: ChannelHandlerContext) {
        println("ProxyOutBoundHandle==")
        ctx.read()
    }

    override fun channelRead0(ctx: ChannelHandlerContext, msg: FullHttpResponse) {
        println("ProxyOutBoundHandle<<<$msg")
        val body = msg.content().toString(CharsetUtil.UTF_8)
        println("ProxyOutBoundHandle body<<<$body")
//        val proxyResponse = try {
//            JSON.parseObject(body, Response::class.java)
//        } catch (e: Exception) {
//            e.printStackTrace()
//            msg.retain()
//            inboundChannel.writeAndFlush(msg).addListener(object : ChannelFutureListener {
//                override fun operationComplete(future: ChannelFuture) {
//                    if (future.isSuccess) {
//                        // was able to flush out data, start to read the next chunk
//                        ctx.channel().read()
//                    } else {
//                        future.channel().close()
//                    }
//                }
//            })
//            return
//        }
//        val headers = CombinedHttpHeaders(true)
//        var outBytes:ByteArray? = null
//        if (proxyResponse.error == null) {
//            for (header in proxyResponse.headers!!) {
//                headers[header.key] = header.value
//            }
//            outBytes = Base64.getDecoder().decode(proxyResponse.content)
//        }else{
//            outBytes = proxyResponse.error.toByteArray()
//        }
//        val response = DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
//                HttpResponseStatus.valueOf(proxyResponse.code),
//                Unpooled.wrappedBuffer(outBytes),
//                headers, DefaultHttpHeaders())
        msg.retain()
        inboundChannel.writeAndFlush(msg.content()).addListener(object : ChannelFutureListener {
            override fun operationComplete(future: ChannelFuture) {
                if (future.isSuccess) {
                    // was able to flush out data, start to read the next chunk
                    ctx.channel().read()
                } else {
                    future.channel().close()
                }
            }
        })
    }

    override fun channelInactive(ctx: ChannelHandlerContext) {
        println("ProxyOutBoundHandle!=")
        ProxyInboundHandler.closeOnFlush(inboundChannel)
    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
        cause.printStackTrace()
        ProxyInboundHandler.closeOnFlush(ctx.channel())
    }

}

package cn.wycode.lambda;

import com.aliyun.fc.runtime.Context;
import com.aliyun.fc.runtime.FunctionComputeLogger;
import com.aliyun.fc.runtime.StreamRequestHandler;

import java.io.*;
import java.net.Socket;
import java.net.URL;
import java.nio.charset.StandardCharsets;

public class Lambda implements StreamRequestHandler {


    @Override
    public void handleRequest(InputStream input, OutputStream output, Context context) throws IOException {
        FunctionComputeLogger logger = context.getLogger();
        StringBuilder sb = new StringBuilder();
        byte[] buffer = new byte[1024];
        int n;
        while ((n = input.read(buffer)) != -1) {
            sb.append(new String(buffer, 0, n, StandardCharsets.UTF_8));
        }
        logger.info("handleRequest>>>" + sb.toString());

        String firstLine = new BufferedReader(new StringReader(sb.toString())).readLine();
        String host;
        int port;
        try {
            if (firstLine.startsWith("CONNECT ")) {
                String[] urlArray = firstLine.split(" ")[1].split(":");
                host = urlArray[0];
                port = Integer.parseInt(urlArray[1]);
            } else {
                URL url = new URL(firstLine.split(" ")[1]);
                host = url.getHost();
                port = url.getPort();

                if (port == -1) {
                    if (url.getProtocol().equalsIgnoreCase("http")) {
                        port = 80;
                    } else if (url.getProtocol().equalsIgnoreCase("https")) {
                        port = 443;
                    } else {
                        throw new IllegalArgumentException("端口无法解析");
                    }
                }
            }
        } catch (Exception e) {
            logger.error("url解析错误:" + e.getMessage());
            return;
        }
        logger.info("host->" + host);
        logger.info("port->" + port);

        Socket s = new Socket(host, port);
        if (firstLine.startsWith("CONNECT ")) {
            String response = "HTTP/1.1 200 Connection Established";
            output.write(response.getBytes(StandardCharsets.UTF_8));
            output.flush();
            sb.setLength(0);
            while ((n = input.read(buffer)) != -1) {
                sb.append(new String(buffer, 0, n, StandardCharsets.UTF_8));
            }
            logger.info("handleRequest>>>" + sb.toString());
        }

        OutputStream httpOutputStream = s.getOutputStream();
        httpOutputStream.write(sb.toString().getBytes(StandardCharsets.UTF_8));

        InputStream httpInputStream = s.getInputStream();
        sb.setLength(0);
        while ((n = httpInputStream.read(buffer)) != -1) {
            output.write(buffer, 0, n);
            sb.append(new String(buffer, 0, n, StandardCharsets.UTF_8));
        }
        httpInputStream.close();
        httpOutputStream.close();

        s.close();
        input.close();
        output.close();
        logger.info("handleRequest<<<" + sb.toString());
    }
}