第 242 章 Netty4 HTTP 组件

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-netty4-http</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-netty4-http</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

// Upload a large data stream to the server
from("direct:upstream-call")
        .bean(Helper.class, "prepareStream")
        .to("netty-http:http://localhost:{{port}}/upstream?disableStreamCache=true")
        .log("get ${body}");

// Read a large data stream from the client
from("netty-http:http://0.0.0.0:{{port}}/upstream?disableStreamCache=true")
        .bean(Helper.class, "processStream")
        .to("mock:stream-size");

// Upload a large data stream to the server
from("direct:upstream-call")
        .bean(Helper.class, "prepareStream")
        .to("netty-http:http://localhost:{{port}}/upstream?disableStreamCache=true")
        .log("get ${body}");

// Read a large data stream from the client
from("netty-http:http://0.0.0.0:{{port}}/upstream?disableStreamCache=true")
        .bean(Helper.class, "processStream")
        .to("mock:stream-size");

// Download a large data stream from the server
from("direct:download-call")
        .to("netty-http:http://localhost:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "asyncProcessStream")
        .log("get ${body}");

// Write a large data stream to the client
from("netty-http:http://0.0.0.0:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "prepareStream");

// Download a large data stream from the server
from("direct:download-call")
        .to("netty-http:http://localhost:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "asyncProcessStream")
        .log("get ${body}");

// Write a large data stream to the client
from("netty-http:http://0.0.0.0:{{port}}/downstream?disableStreamCache=true")
        .bean(Helper.class, "prepareStream");

    public static void processStream(Exchange exchange) throws Exception {
        InputStream is = exchange.getIn().getBody(InputStream.class);

        byte[] buffer = new byte[1024];
        long read = 0;
        long total = 0;
        while ((read = is.read(buffer, 0, buffer.length)) != -1) {
            total += read;
        }

        exchange.getIn().setBody(new Long(total));
    }

    public static CompletableFuture<Void> asyncProcessStream(Exchange exchange) {
        return CompletableFuture.runAsync(() -> {
            try {
                processStream(exchange);
            } catch (Exception e) {
                exchange.setException(e);
            }
        });
    }

    public static void processStream(Exchange exchange) throws Exception {
        InputStream is = exchange.getIn().getBody(InputStream.class);

        byte[] buffer = new byte[1024];
        long read = 0;
        long total = 0;
        while ((read = is.read(buffer, 0, buffer.length)) != -1) {
            total += read;
        }

        exchange.getIn().setBody(new Long(total));
    }

    public static CompletableFuture<Void> asyncProcessStream(Exchange exchange) {
        return CompletableFuture.runAsync(() -> {
            try {
                processStream(exchange);
            } catch (Exception e) {
                exchange.setException(e);
            }
        });
    }