Node.js 中的 HTTP/2 POST 请求

Question

如何在 Node.js 中进行 HTTP/2 以下的 post 请求？

curl --http2 "POST" "http://hostname:8088/query-stream" -d $'{"sql": "SELECT * FROM `USERPROFILE` EMIT CHANGES;", "properties": {"ksql.streams.auto.offset.reset": "earliest" } }'
enter code here

您为 HTTP/2 请求提供什么 Node.js 模块？

Answer 1

我已经解决了这个问题。如果有人需要拐杖使用以下代码

const http2 = require('http2');
const http2PostRequest = (url, path, data) => new Promise((resolve, reject) => {
    console.log('url => ', url);
    console.log('data => ', data);

    const client = http2.connect(url);
    console.log(' http2.constants.HTTP2_METHOD_POST => ', http2.constants.HTTP2_METHOD_POST)
    console.log(' http2.constants.HTTP2_PATH => ', `${path}`);

    const req = client.request({
        [http2.constants.HTTP2_HEADER_SCHEME]: "https",
        'Content-Type': 'application/json',
        'Content-Length': data.length,
        ":method": "POST",
        ":path": path
    });
    req.setEncoding('utf8');
    req.on('response', (headers, flags) => {
        console.log('headers => ', headers);
        console.log('flags => ', flags);

        req.on('data', d => {
            console.log('data => ', JSON.parse(d));
        })

        req.on('close', () => {
            try {
                resolve(data);
            } catch (e) {
                resolve(e);
            }
        })
        req.on('error', error => {
            reject(error);
        })
    });
    req.write(data);
    req.end();
});

const data = {
    "sql": "SELECT * FROM `FOO_02` EMIT CHANGES;",
    "properties": {"ksql.streams.auto.offset.reset": "earliest"}
};
http2PostRequest(
    "http://localhost:8088",
    "/query-stream",
    JSON.stringify(data)
).then((res) => {
    console.log('res => ', res);
})

Answer 2

感谢您的回答，但我的问题是连接到 ksqDB 中的 Kafka 流，而不是连接 Kafka 主题。我想做一个 http2 请求，比如

curl --http2 "POST" "hostname:8088/query-stream" -d $'{"sql": "SELECT

• FROM USERPROFILE EMIT CHANGES;", "properties": {"ksql.streams.auto.offset.reset": "earliest" } }'

Answer 3

这是一个通过发布者和侦听器函数将其作为服务处理的类。

export class KafkaService {

    opts = { brokers: 'localhost:9092', groupId: '0' };
    kafkaObservable = require('kafka-observable');
    static instance = null;
    static getInstance() {
        if(!KafkaService.instance){
            KafkaService.instance = new KafkaService();
        }
        return KafkaService.instance;
    }

    publishMessage(topic, messages) {
        this.kafkaObservable.toTopic(topic, messages, this.opts).subscribe(x => x);
    }

    consumerListener(topic) {
        return this.kafkaObservable.fromTopic(topic, this.opts)
        .map(({message}) => message.value.toString('utf8'));;
    }

}

在您的应用程序中，您可以使用以下代码发送和接收消息

import {ItemBet} from "./src/handlers/item-bet.js";
import express from "express"
import { KafkaService } from "./src/services/kafka-service.js";

var app = express();
var server = require('http').Server(app);
var io = require('socket.io')(server);

app.use(express.static('./public'));

const kafka = KafkaService.getInstance();

io.on('connection', function(socket){    
    console.log("Someone is been connected in a socket");

    socket.on('purchase-message', function(data){
        console.log('Shopping Card, purchase-message', data);
        console.log('Shopping Card, kafka-producer-message');
        console.log('Shopping Card, SOCKET', socket.id);

        kafka.publishMessage('proceed-transaction', [{socketId: socket.id, data: data}]);     
    });
});


kafka.consumerListener('transaction-response')
.map( res => JSON.parse(res))
.subscribe(res => {        
    console.log('Shopping Card, kafka-customer, transaction-response', res);
    io.to(res.socketId).emit('transaction-response', res.data);
});

server.listen(1024);