1102 lines
18 KiB
Markdown
1102 lines
18 KiB
Markdown
|
||
|
||
# 技术选型
|
||
### 串口数据接收
|
||
1. 串口模拟器
|
||
1. socat
|
||
1. **Create Virtual Serial Port Pair** Use `socat` to create two linked virtual serial ports. This example creates two virtual serial ports: `/dev/ttyV0` and `/dev/ttyV1`.
|
||
```bash
|
||
socat -d -d PTY,link=./dev/ttyV0,raw,echo=0 PTY,link=./dev/ttyV1,raw,echo=0
|
||
```
|
||
**Explanation:**
|
||
- `-d -d`: Enables debug output to see the details of the operations.
|
||
- `PTY,link=/dev/ttyV0,raw,echo=0`: Creates a pseudo-terminal (PTY) with a symbolic link `/dev/ttyV0`, in raw mode with no echo.
|
||
- `PTY,link=/dev/ttyV1,raw,echo=0`: Creates another PTY with a symbolic link `/dev/ttyV1`, in raw mode with no echo.
|
||
-
|
||
```bash
|
||
screen ./dev/ttyV0 9600
|
||
|
||
echo "hello" > ./dev/ttyV1
|
||
```
|
||
### 消息队列
|
||
nats-server
|
||
|
||
|
||
# 架构设计
|
||
|
||
micrioprofile
|
||
openliberty
|
||
|
||
|
||
micronaut
|
||
|
||
|
||
1. 串口读取
|
||
2. 报文拆分
|
||
3. 动态解析
|
||
4. 计划解析
|
||
5. 总线消息生成
|
||
6. 动态信息转发
|
||
7. 计划转发
|
||
|
||
串口 => 消息流原始 => 已分离电报消息 => 待发送 => 消息总线
|
||
|
||
### 系统总线适配
|
||
日计划处理
|
||
季度计划处理
|
||
航班主要信息同步
|
||
动态消息转发
|
||
### 串口转消息队列
|
||
命令行工具
|
||
1. 串口参数
|
||
2. 消息队列参数
|
||
|
||
### 电报存储
|
||
postgresql
|
||
pocketbase
|
||
|
||
nats
|
||
nui
|
||
pocketbase
|
||
meilisearch
|
||
|
||
compose.yml
|
||
```yaml
|
||
services:
|
||
nats:
|
||
container_name: nats
|
||
image: nats
|
||
ports:
|
||
- "4222:4222"
|
||
- "6222:6222"
|
||
- "8222:8222"
|
||
restart: unless-stopped
|
||
|
||
nui:
|
||
container_name: nui
|
||
image: ghcr.io/nats-nui/nui
|
||
volumes:
|
||
- ./db:/db
|
||
- ./creds:/nats-creds:ro
|
||
ports:
|
||
- "31311:31311"
|
||
restart: unless-stopped
|
||
|
||
pocketbase:
|
||
image: ghcr.io/muchobien/pocketbase:latest
|
||
container_name: pocketbase
|
||
restart: unless-stopped
|
||
command:
|
||
#- --encryptionEnv #optional
|
||
#- ENCRYPTION #optional
|
||
environment:
|
||
ENCRYPTION: example #optional
|
||
ports:
|
||
- "8090:8090"
|
||
volumes:
|
||
- ./data:/pb_data
|
||
- ./public:/pb_public #optional
|
||
- ./hooks:/pb_hooks #optional
|
||
|
||
healthcheck: #optional (recommended) since v0.10.0
|
||
test: wget --no-verbose --tries=1 --spider http://localhost:8090/api/health || exit 1
|
||
interval: 30s
|
||
timeout: 5s
|
||
retries: 5
|
||
meilisearch:
|
||
container_name: meilisearch
|
||
image: getmeili/meilisearch:v1.8
|
||
environment:
|
||
- http_proxy
|
||
- https_proxy
|
||
- MEILI_MASTER_KEY=${MEILI_MASTER_KEY:-masterKey}
|
||
- MEILI_NO_ANALYTICS=${MEILI_NO_ANALYTICS:-true}
|
||
- MEILI_ENV=${MEILI_ENV:-development}
|
||
ports:
|
||
- ${MEILI_PORT:-7700}:7700
|
||
volumes:
|
||
- ./meili_data:/meili_data
|
||
restart: unless-stopped
|
||
```
|
||
|
||
|
||
nat with monitor:
|
||
```docker-compose.yml
|
||
networks:
|
||
monitor-net:
|
||
driver: bridge
|
||
|
||
services:
|
||
nats:
|
||
image: nats
|
||
container_name: nats
|
||
restart: always
|
||
command: -c /etc/nats/nats.conf
|
||
ports:
|
||
- "4222:4222" # client port
|
||
- "6222:6222" # cluster port
|
||
- "8222:8222" # monitoring port
|
||
volumes:
|
||
- ./nats.conf:/etc/nats/nats.conf
|
||
- $JETSTREAM_STORAGE:/data
|
||
networks:
|
||
- monitor-net
|
||
|
||
exporter:
|
||
image: natsio/prometheus-nats-exporter
|
||
container_name: nats-exporter
|
||
restart: always
|
||
command:
|
||
# see https://github.com/nats-io/prometheus-nats-exporter/blob/main/main.go#L87
|
||
# "-connz", # connection metrics
|
||
- -connz_detailed # advanced connection metrics
|
||
- -jsz
|
||
- all # jetstream metrics
|
||
- -routez # route metrics
|
||
- -subz # subscription metrics
|
||
- -varz # general metrics
|
||
- -prefix=nats # prefix for all metrics
|
||
- -use_internal_server_id # using serverID from /varz
|
||
- http://nats:8222/
|
||
networks:
|
||
- monitor-net
|
||
labels:
|
||
org.label-schema.group: "nats-monitoring"
|
||
depends_on:
|
||
- nats
|
||
|
||
# ports:
|
||
# - "7777:7777"
|
||
|
||
surveyor:
|
||
image: natsio/nats-surveyor
|
||
container_name: nats-surveyor
|
||
restart: always
|
||
volumes:
|
||
- ./observations:/observations
|
||
- $JETSTREAM_STORAGE:/data
|
||
command: |
|
||
-s "${NATS_SURVEYOR_SERVERS}" --accounts --observe /observations --jetstream /data
|
||
networks:
|
||
- monitor-net
|
||
labels:
|
||
org.label-schema.group: "nats-monitoring"
|
||
depends_on:
|
||
- nats
|
||
|
||
prometheus:
|
||
image: prom/prometheus
|
||
container_name: prometheus
|
||
restart: always
|
||
volumes:
|
||
- ./prometheus/:/etc/prometheus/
|
||
- $PROMETHEUS_STORAGE:/usr/local/share/prometheus
|
||
command: --config.file=/etc/prometheus/prometheus.yml --storage.tsdb.path=/usr/local/share/prometheus
|
||
networks:
|
||
- monitor-net
|
||
labels:
|
||
org.label-schema.group: "nats-monitoring"
|
||
ports:
|
||
- "9090:9090"
|
||
depends_on:
|
||
- surveyor
|
||
- exporter
|
||
|
||
grafana:
|
||
image: grafana/grafana
|
||
container_name: grafana
|
||
restart: always
|
||
ports:
|
||
- "3000:3000"
|
||
volumes:
|
||
- ./grafana/dashboards:/var/lib/grafana/dashboards
|
||
- ./grafana/provisioning:/etc/grafana/provisioning
|
||
networks:
|
||
- monitor-net
|
||
labels:
|
||
org.label-schema.group: "nats-monitoring"
|
||
depends_on:
|
||
- prometheus
|
||
```
|
||
|
||
```dash
|
||
```
|
||
|
||
### 电报解析，电报存储，电报转发
|
||
管理界面接口
|
||
|
||
### 管理界面
|
||
电报界面
|
||
日计划界面
|
||
全文检索
|
||
typesense
|
||
meli
|
||
|
||
nextjs shadcn
|
||
svelte
|
||
shadcn
|
||
melt-ui: https://melt-ui.com
|
||
flowbite: [Flowbite Svelte](https://flowbite-svelte.com/)
|
||
|
||
htmx:
|
||
go:
|
||
chi htmx tailwindcss
|
||
echo htmx tailwindcss
|
||
python:
|
||
|
||
deno:
|
||
fresh
|
||
|
||
|
||
Telegram Message Body Expression:
|
||
|
||
FPL:
|
||
```regex
|
||
|
||
\((?P<category>[A-Z]{3})-(?P<number>[A-Z]+\d+)-(?P<indicator>[A-Z]{2})\n-(?P<aircraft>[A-Z]+\d+\/?[A-Z]?)\n?-(?P<surve>.*)\n?-(?P<departure>[A-Z]{4})(?P<departure_time>\d{4})\n?-(?P<speed>[A-Z]+\d+)(?P<level>[A-Z0-9]+)\s+(?P<route>(.|\n)+)\n-(?P<destination>[A-Z]{4})(?P<estt>\d{4})\s+(?P<alter>[A-Z]{4})\n?-([A-Z]{3}\/(?:[A-Z]{4}\d{4}\s?)+)?(?P<other>(?m)[A-Z]{3}\/(.|\n)*)\)$
|
||
|
||
```
|
||
|
||
ARR:
|
||
```regex
|
||
|
||
\\((?P<type>[A-Z]{3})\\-(?P<number>([A-Z]+\\d+))\\/?(?P<ssr>[A-Z]+\\d+)\\-(?P<departure>[A-Z]{4})\\-(?P<arrival>[A-Z]{4})(?P<time>\\d{4})\\)
|
||
|
||
```
|
||
|
||
DEP:
|
||
```regex
|
||
|
||
^\\((?P<type>[A-Z]{3})\\-(?P<number>([A-Z]+\\d+))\\/?(?P<ssr>[A-Z]+\\d+)\\-(?P<departure>[A-Z]{4})(?P<departure_time>\\d{4})\\-(?P<destination>[A-Z]{4})(?P<destination_time>\\d{4})?\\-?(?P<alter>[A-Z]{4})?\\-?(?P<other>.*)?\\)$
|
||
```
|
||
|
||
NEW FPL:
|
||
```regex
|
||
\((?P<category>[A-Z]{3})-(?P<number>[A-Z]+\d+)-(?P<indicator>[A-Z]{2})\n-(?P<aircraft>[A-Z]+\d+\/?[A-Z]?)\n?-(?P<surve>.*)\n?-(?P<departure>[A-Z]{4})(?P<departure_time>\d{4})\n?-(?P<speed>[A-Z]+\d+)(?P<level>[A-Z0-9]+)\s+(?P<route>(.|\n)+)\n-(?P<destination>[A-Z]{4})(?P<estt>\d{4})\s?(?P<alter>(\s[A-Z]{4})+)\n?-([A-Z]{3}\/(?:[A-Z]{4}\d{4}\s?)+)?(?P<other>(?m)[A-Z]{3}\/(.|\n)*)\)$
|
||
```
|
||
|
||
CNL:
|
||
```regex
|
||
^\((?P<category>[A-Z]{3})-(?P<number>\w+\d+)-?(?P<departure>[A-Z]{4})?-?(?<destination>[A-Z]{4})\)$
|
||
```
|
||
|
||
DLA:
|
||
```regex
|
||
^\((?P<category>[A-Z]{3})-(?P<number>\w+\d+)-?(?P<departure>[A-Z]{4})(?P<departure_time>\d{4})?-?(?<destination>[A-Z]{4})(?<destination_time>\d{4})?\)$
|
||
```
|
||
|
||
|
||
|
||
```CA
|
||
ZCZC TMQ1067 150518
|
||
|
||
|
||
QU TSNZPCA
|
||
|
||
.
|
||
|
||
QU SZXZPCA TAOZPCA WNZZPCA SHATZCR PEKZPCA WUHZPCA TSNZPCA
|
||
|
||
.TNAUOSC 150519
|
||
|
||
PLN 16MAY
|
||
|
||
01) SC4731/4732/4667/4668 B2968/B733 ILS I (7) TAO/2340 WNZ/0225
|
||
|
||
SZX/
|
||
|
||
0500 WNZ/0725 TAO/1000 SHA/1210
|
||
|
||
SI:AWY/TAO WNZ SZX WNZ TAO SHA TAO
|
||
|
||
02) SC4617/4618/4655/4656/4087/4088/4669/4670 B2996/B733 ILS I (8)
|
||
|
||
TA
|
||
|
||
O/0030 WUH/0255 TAO/0520 PEK/0730 TAO/0945 ICN/1155 TAO/1350
|
||
|
||
SHA/1550
|
||
|
||
SI:AWY/TAO WUH TAO PEK TAO ICN TAO SHA TAO
|
||
|
||
03) SC4621/4622/4733/4734 B3005/CRJ2 ILS I (5) TAO/0005 DLC/0130
|
||
|
||
TAO/
|
||
|
||
0300 XIY/0545 KWE/0745 XIY/0950
|
||
|
||
SI:AWY/TAO DLC TAO XIY KWE XIY TAO
|
||
|
||
04) SC4607/4608/4623/4880/4879/4624 B3007/CRJ2 ILS I (5) TAO/2355
|
||
|
||
TY
|
||
|
||
N/0150 ZGC/0350 TYN/0545 TAO/0745 DLC/0900 YNT/1015 HGH/1225
|
||
|
||
YNT/1435 DLC
|
||
|
||
/1550
|
||
|
||
SI:AWY/TAO TYN ZGC TYN TAO DLC YNT HGH YNT DLC TAO
|
||
|
||
05) SC4679/4680/4711/4712 B3079/CRJ7 ILS I (5) TAO/0001 HGH/0210
|
||
|
||
FOC/
|
||
|
||
0350 HGH/0530 TAO/0735 HGH/0940 NNG/1255 HGH/1525
|
||
|
||
SI:AWY/TAO HGH FOC HGH TAO HGH NNG HGH TAO
|
||
|
||
06) SC4717/4817/4818/4718 B3080/CRJ7 ILS I (5) TAO/0005 TSN/0135
|
||
|
||
HET/
|
||
|
||
0320 TGO/0535 HET/0750 TSN/0930
|
||
|
||
SI:AWY/TAO TSN HET TGO HET TSN TAO
|
||
|
||
07) SC4709/4981/4982 B5065/B733 ILS I (7) TAO/2355 NGB/0155
|
||
|
||
XMN/0430
|
||
|
||
TNA/0725 HET/0950 TNA/1200
|
||
|
||
SI:AWY/TAO NGB XMN TNA HET TNA XMN
|
||
|
||
08) SC4651/1152/1155/1156/1165/1166/1167/1168 B5205/B737 ILS I (7)
|
||
|
||
TA
|
||
|
||
O/2340 PEK/0140 TNA/0320 PEK/0510 TNA/0715 SHA/0925 TNA/1135
|
||
|
||
SHA/1340
|
||
|
||
SI:AWY/TAO PEK TNA PEK TNA SHA TNA SHA TNA
|
||
|
||
09) SC4695/4696/4657/4658 B5331/B738 ILS I (8) TAO/2350 NKG/0145
|
||
|
||
KMG/
|
||
|
||
0505 NKG/0830 TAO/1045 PEK/1300
|
||
|
||
SI:AWY/TAO NKG KMG NKG TAO PEK TAO
|
||
|
||
10) SC4601/4602/4675/4672 B5348/B738 ILS I (10) TAO/2330 TNA/0055
|
||
|
||
CTU
|
||
|
||
/0405 TNA/0645 TAO/0830 CAN/1225 LYI/1515
|
||
|
||
SI:AWY/TAO TNA CTU TNA TAO CAN LYI TAO
|
||
|
||
11) SC4705/4706/4751/4752 B5349/B738 ILS I (8) TAO/2340 HFE/0135
|
||
|
||
HHA/
|
||
|
||
0335 HFE/0530 TAO/0720 CGO/0930 KWL/1210 CGO/1440
|
||
|
||
SI:AWY/TAO HFE HHA HFE TAO CGO KWL CGO TAO
|
||
|
||
12) SC4671/4676/4665/4666/4659/4660 B5350/B738 ILS I (8) TAO/2350
|
||
|
||
LYI
|
||
|
||
/0105 CAN/0415 TAO/0745 SHA/0950 TAO/1155 PEK/1400
|
||
|
||
SI:AWY/TAO LYI CAN TAO SHA TAO
|
||
|
||
PART ONE CONTINUED
|
||
|
||
=
|
||
|
||
|
||
|
||
|
||
NNNN
|
||
|
||
```
|
||
|
||
|
||
```fpl
|
||
ZCZC TMQ2544 141652
|
||
|
||
|
||
FF ZBTJZXZX
|
||
|
||
|
||
141652 ZBTJZPZX
|
||
|
||
|
||
(FPL-JAE7433-IS
|
||
|
||
|
||
-B744/H-SXIRPZJWY/S
|
||
|
||
|
||
-ZBTJ1755
|
||
|
||
|
||
-K0926S0920
|
||
|
||
|
||
-EDDF0948 EDDK
|
||
|
||
|
||
-EET/ZMUB0100 UNKL0236 UNWW0332 UNNT0332 USRR0447 USHH0507
|
||
|
||
|
||
USSS0535 UUYY0602 ULKK0634 ULWW0653 ULLL0720 EETT0748 EVRR0815
|
||
|
||
|
||
ESAA0821 EPWW0848 EDUU0900
|
||
|
||
|
||
REG/B2422 SEL/JLAD OPR/JADE CARGO DAT/S RVR/200
|
||
|
||
|
||
NAV/RNAV1 RNAV5 RNP4
|
||
|
||
|
||
RMK/AGCS EQUIPPED
|
||
|
||
|
||
ACARS EQUIPPED/TCAS EQUIPPED/FOREIGN PILOT
|
||
|
||
|
||
E/1148 P/TBN R/UV S/M J/LF D/1 15 C YELLOW
|
||
|
||
|
||
A/WHITE GREEN)
|
||
|
||
NNNN
|
||
|
||
```
|
||
|
||
|
||
计划：
|
||
|
||
airway：
|
||
```regex
|
||
(?P<arr_time>\d{4}(\(\d{2}\w{3}\))?)(?P<airport>\w{3})\/?(?P<dep_time>\d{4}(\(\d{2}\w{3}\))?)
|
||
```
|
||
|
||
|
||
## Store
|
||
|
||
nats:
|
||
serial to nats
|
||
```yaml
|
||
services:
|
||
nats:
|
||
container_name: nats
|
||
image: nats
|
||
ports:
|
||
- "4222:4222"
|
||
- "6222:6222"
|
||
- "8222:8222"
|
||
command: >
|
||
-js
|
||
-D
|
||
environment:
|
||
- NATS_SERVER_LOG_NAME=nats-server
|
||
volumes:
|
||
- ./nats_data:/data # Persistent storage for JetStream
|
||
restart: unless-stopped
|
||
|
||
nui:
|
||
container_name: nui
|
||
image: ghcr.io/nats-nui/nui
|
||
volumes:
|
||
- ./db:/db
|
||
- ./creds:/nats-creds:ro
|
||
ports:
|
||
- "31311:31311"
|
||
restart: unless-stopped
|
||
```
|
||
|
||
|
||
|
||
nats jetstream:
|
||
```go
|
||
package main
|
||
|
||
|
||
|
||
import (
|
||
|
||
"fmt"
|
||
|
||
"log"
|
||
|
||
"time"
|
||
|
||
|
||
|
||
"github.com/nats-io/nats.go"
|
||
|
||
)
|
||
|
||
|
||
|
||
func publishMessage(js nats.JetStreamContext, subject string, message []byte) error {
|
||
|
||
ack, err := js.Publish(subject, message)
|
||
|
||
if err != nil {
|
||
|
||
return fmt.Errorf("error publishing message: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
log.Printf("Published message to %s, stream sequence: %d", subject, ack.Sequence)
|
||
|
||
return nil
|
||
|
||
}
|
||
|
||
|
||
|
||
func publishMessages(js nats.JetStreamContext, subject string, count int) error {
|
||
|
||
for i := 1; i <= count; i++ {
|
||
|
||
msg := []byte(fmt.Sprintf("Hello JetStream! Message #%d", i))
|
||
|
||
|
||
|
||
if err := publishMessage(js, subject, msg); err != nil {
|
||
|
||
return fmt.Errorf("failed to publish message %d: %v", i, err)
|
||
|
||
}
|
||
|
||
|
||
|
||
time.Sleep(1 * time.Second)
|
||
|
||
}
|
||
|
||
return nil
|
||
|
||
}
|
||
|
||
|
||
|
||
func setupStream(js nats.JetStreamContext, name string, subjects []string) error {
|
||
|
||
streamConfig := &nats.StreamConfig{
|
||
|
||
Name: name,
|
||
|
||
Subjects: subjects,
|
||
|
||
Storage: nats.MemoryStorage,
|
||
|
||
MaxAge: time.Hour,
|
||
|
||
}
|
||
|
||
|
||
|
||
// Try to create stream, if it already exists, update it
|
||
|
||
_, err := js.AddStream(streamConfig)
|
||
|
||
if err != nil {
|
||
|
||
if err == nats.ErrStreamNameAlreadyInUse {
|
||
|
||
_, err = js.UpdateStream(streamConfig)
|
||
|
||
if err != nil {
|
||
|
||
return fmt.Errorf("error updating stream: %v", err)
|
||
|
||
}
|
||
|
||
log.Printf("Updated stream %s", name)
|
||
|
||
return nil
|
||
|
||
}
|
||
|
||
return fmt.Errorf("error creating stream: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
log.Printf("Created stream %s", name)
|
||
|
||
return nil
|
||
|
||
}
|
||
|
||
|
||
|
||
func connectToNATS() (*nats.Conn, nats.JetStreamContext, error) {
|
||
|
||
opts := []nats.Option{
|
||
|
||
nats.Timeout(5 * time.Second),
|
||
|
||
nats.RetryOnFailedConnect(true),
|
||
|
||
nats.MaxReconnects(5),
|
||
|
||
nats.ReconnectWait(time.Second),
|
||
|
||
nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
|
||
|
||
log.Printf("Disconnected due to: %s, will attempt reconnects for 5 times", err)
|
||
|
||
}),
|
||
|
||
nats.ReconnectHandler(func(nc *nats.Conn) {
|
||
|
||
log.Printf("Reconnected [%s]", nc.ConnectedUrl())
|
||
|
||
}),
|
||
|
||
nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
|
||
|
||
log.Printf("Error: %v", err)
|
||
|
||
}),
|
||
|
||
}
|
||
|
||
|
||
|
||
nc, err := nats.Connect("nats://localhost:4222", opts...)
|
||
|
||
if err != nil {
|
||
|
||
return nil, nil, fmt.Errorf("error connecting to NATS: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
js, err := nc.JetStream()
|
||
|
||
if err != nil {
|
||
|
||
nc.Close()
|
||
|
||
return nil, nil, fmt.Errorf("error getting JetStream context: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
// Setup the stream
|
||
|
||
if err := setupStream(js, "TEST_STREAM", []string{"test.>"}); err != nil {
|
||
|
||
nc.Close()
|
||
|
||
return nil, nil, err
|
||
|
||
}
|
||
|
||
|
||
|
||
return nc, js, nil
|
||
|
||
}
|
||
|
||
|
||
|
||
func subscribeToMessages(js nats.JetStreamContext, subject string, expectedMsgs int) (*nats.Subscription, chan bool, error) {
|
||
|
||
done := make(chan bool)
|
||
|
||
msgCount := 0
|
||
|
||
|
||
|
||
// Create a JetStream subscription
|
||
|
||
sub, err := js.Subscribe(subject, func(msg *nats.Msg) {
|
||
|
||
// Handle the message
|
||
|
||
log.Printf("Received message: %s\n", string(msg.Data))
|
||
|
||
log.Printf("Subject: %s\n", msg.Subject)
|
||
|
||
|
||
|
||
// Get metadata if available
|
||
|
||
if meta, err := msg.Metadata(); err == nil {
|
||
|
||
log.Printf("Sequence: %d\n", meta.Sequence.Stream)
|
||
|
||
log.Printf("Timestamp: %v\n", meta.Timestamp)
|
||
|
||
}
|
||
|
||
|
||
|
||
// Acknowledge the message
|
||
|
||
if err := msg.Ack(); err != nil {
|
||
|
||
log.Printf("Error acknowledging message: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
// Increment message count and check if we're done
|
||
|
||
msgCount++
|
||
|
||
if msgCount >= expectedMsgs {
|
||
|
||
done <- true
|
||
|
||
}
|
||
|
||
}, nats.DeliverAll(), nats.AckExplicit())
|
||
|
||
|
||
|
||
if err != nil {
|
||
|
||
return nil, nil, fmt.Errorf("error subscribing to subject %s: %v", subject, err)
|
||
|
||
}
|
||
|
||
|
||
|
||
log.Printf("Subscribed to %s", subject)
|
||
|
||
return sub, done, nil
|
||
|
||
}
|
||
|
||
|
||
|
||
func main() {
|
||
|
||
// Connect to NATS and get JetStream context
|
||
|
||
nc, js, err := connectToNATS()
|
||
|
||
if err != nil {
|
||
|
||
log.Fatalf("Failed to connect: %v", err)
|
||
|
||
}
|
||
|
||
defer nc.Close()
|
||
|
||
|
||
|
||
expectedMsgs := 5
|
||
|
||
|
||
|
||
// Subscribe to messages
|
||
|
||
sub, done, err := subscribeToMessages(js, "test.>", expectedMsgs)
|
||
|
||
if err != nil {
|
||
|
||
log.Fatalf("Failed to subscribe: %v", err)
|
||
|
||
}
|
||
|
||
defer sub.Unsubscribe()
|
||
|
||
|
||
|
||
// Publish messages
|
||
|
||
if err := publishMessages(js, "test.message", expectedMsgs); err != nil {
|
||
|
||
log.Printf("Error in publish sequence: %v", err)
|
||
|
||
}
|
||
|
||
|
||
|
||
// Wait for all messages to be received
|
||
|
||
log.Printf("Waiting for %d messages...", expectedMsgs)
|
||
|
||
<-done
|
||
|
||
log.Printf("All %d messages received. Exiting...", expectedMsgs)
|
||
|
||
}
|
||
```
|
||
|
||
|
||
|
||
|
||
nats with solidjs
|
||
|
||
main.ts:
|
||
```typescript
|
||
import { Hono } from "jsr:@hono/hono";
|
||
|
||
import { connect, StringCodec } from "nats";
|
||
|
||
import process from "node:process";
|
||
|
||
import { upgradeWebSocket } from "jsr:@hono/hono/deno";
|
||
|
||
|
||
|
||
// Connect to NATS Server
|
||
|
||
let nc: Awaited<ReturnType<typeof connect>>;
|
||
|
||
try {
|
||
|
||
nc = await connect({ servers: "nats://localhost:4222" });
|
||
|
||
console.log("✅ Connected to NATS");
|
||
|
||
} catch (error) {
|
||
|
||
console.error("❌ Failed to connect to NATS:", error);
|
||
|
||
process.exit(1);
|
||
|
||
}
|
||
|
||
const sc = StringCodec();
|
||
|
||
|
||
|
||
// Store active WebSocket clients
|
||
|
||
const clients = new Set<WebSocket>();
|
||
|
||
|
||
|
||
// Create Hono app
|
||
|
||
const app = new Hono();
|
||
|
||
|
||
|
||
// WebSocket Upgrade Route
|
||
|
||
app.get("/ws", upgradeWebSocket((c) => {
|
||
|
||
return {
|
||
|
||
onMessage(event, ws) {
|
||
|
||
console.log(`📩 Received from client: ${event.data}`);
|
||
|
||
// Broadcast to other clients
|
||
|
||
for (const client of clients) {
|
||
|
||
if (client !== ws && client.readyState === WebSocket.OPEN) {
|
||
|
||
client.send(event.data);
|
||
|
||
}
|
||
|
||
}
|
||
|
||
// Publish to NATS
|
||
|
||
nc.publish("updates", sc.encode(event.data));
|
||
|
||
},
|
||
|
||
onClose() {
|
||
|
||
console.log("❌ WebSocket client disconnected");
|
||
|
||
},
|
||
|
||
onOpen(ws) {
|
||
|
||
clients.add(ws);
|
||
|
||
console.log("🔗 New WebSocket client connected");
|
||
|
||
}
|
||
|
||
};
|
||
|
||
}));
|
||
|
||
|
||
|
||
// Start WebSocket Server
|
||
|
||
console.log("🚀 Starting Hono WebSocket Server on http://localhost:3001/ws");
|
||
|
||
|
||
|
||
// Subscribe to NATS topic and forward messages to WebSockets
|
||
|
||
(async () => {
|
||
|
||
if (!nc) return;
|
||
|
||
const sub = nc.subscribe("updates");
|
||
|
||
for await (const msg of sub) {
|
||
|
||
const message = sc.decode(msg.data);
|
||
|
||
console.log(`📩 Received from NATS: ${message}`);
|
||
|
||
|
||
|
||
for (const client of clients) {
|
||
|
||
if (client.readyState === WebSocket.OPEN) {
|
||
|
||
client.send(message);
|
||
|
||
}
|
||
|
||
}
|
||
|
||
}
|
||
|
||
})();
|
||
|
||
|
||
|
||
// Start the server
|
||
|
||
await Deno.serve({ port: 3001 }, app.fetch);
|
||
```
|
||
|
||
|
||
websocketclient.tsx
|
||
|
||
```tsx
|
||
import { createSignal, onCleanup, onMount } from "solid-js";
|
||
|
||
|
||
|
||
export default function WebSocketClient() {
|
||
|
||
const [messages, setMessages] = createSignal<string[]>([]);
|
||
|
||
const [input, setInput] = createSignal("");
|
||
|
||
let ws: WebSocket;
|
||
|
||
|
||
|
||
// Establish WebSocket connection
|
||
|
||
onMount(() => {
|
||
|
||
ws = new WebSocket("ws://localhost:3001/ws");
|
||
|
||
|
||
|
||
ws.onopen = () => console.log("✅ WebSocket connected");
|
||
|
||
ws.onmessage = (event) => {
|
||
|
||
console.log("Received message:", event.data);
|
||
|
||
setMessages((prev) => [...prev, event.data]);
|
||
|
||
};
|
||
|
||
|
||
|
||
ws.onclose = () => console.log("❌ WebSocket disconnected");
|
||
|
||
ws.onerror = (err) => console.error("❗ WebSocket error:", err);
|
||
|
||
|
||
|
||
// Cleanup WebSocket connection when component unmounts
|
||
|
||
onCleanup(() => ws.close());
|
||
|
||
});
|
||
|
||
|
||
|
||
// Send message via WebSocket (which then publishes to NATS)
|
||
|
||
const sendMessage = () => {
|
||
|
||
if (ws.readyState === WebSocket.OPEN && input().trim() !== "") {
|
||
|
||
ws.send(input());
|
||
|
||
setMessages((prev) => [...prev, input()]);
|
||
|
||
setInput(""); // Clear input after sending
|
||
|
||
}
|
||
|
||
};
|
||
|
||
|
||
|
||
return (
|
||
|
||
<div>
|
||
|
||
<h2>📡 NATS WebSocket Chat</h2>
|
||
|
||
|
||
|
||
<div>
|
||
|
||
<input
|
||
|
||
type="text"
|
||
|
||
placeholder="Type a message..."
|
||
|
||
value={input()}
|
||
|
||
onInput={(e) => setInput(e.currentTarget.value)}
|
||
|
||
/>
|
||
|
||
<button onClick={sendMessage}>Send</button>
|
||
|
||
</div>
|
||
|
||
|
||
|
||
<ul>
|
||
|
||
{messages().map((msg, index) => (
|
||
|
||
<li>
|
||
|
||
{index} {msg}
|
||
|
||
</li>
|
||
|
||
))}
|
||
|
||
</ul>
|
||
|
||
</div>
|
||
|
||
);
|
||
|
||
}
|
||
```
|
||
|
||
|