From aa3ea917451727e17183edea85bc360ed21a28f0 Mon Sep 17 00:00:00 2001 From: John Ajera Date: Tue, 18 Aug 2026 10:02:52 +1200 Subject: [PATCH] feat: add ringserver-feeder image SeedLink client that fans each packet to every ringserver DataLink target with the same packet ID (WRITE flag I). Includes compose lab and CI matching ringserver. --- .dockerignore | 9 + .github/dependabot.yml | 22 + .github/workflows/auto-merge.yml | 17 + .github/workflows/build-and-release.yml | 17 + .github/workflows/ci.yml | 14 + .github/workflows/commitmsg-conform.yml | 14 + .github/workflows/markdown-lint.yml | 14 + Dockerfile | 60 +++ LICENSE | 21 + README.md | 51 +- lab/config/replica.conf | 3 + lab/config/upstream.conf | 3 + lab/docker-compose.yml | 45 ++ lab/miniseed/test.mseed | Bin 0 -> 4096 bytes lab/prove.sh | 50 ++ src/Makefile | 18 + src/ringserver-feeder.c | 653 ++++++++++++++++++++++++ 17 files changed, 1010 insertions(+), 1 deletion(-) create mode 100644 .dockerignore create mode 100644 .github/dependabot.yml create mode 100644 .github/workflows/auto-merge.yml create mode 100644 .github/workflows/build-and-release.yml create mode 100644 .github/workflows/ci.yml create mode 100644 .github/workflows/commitmsg-conform.yml create mode 100644 .github/workflows/markdown-lint.yml create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 lab/config/replica.conf create mode 100644 lab/config/upstream.conf create mode 100644 lab/docker-compose.yml create mode 100644 lab/miniseed/test.mseed create mode 100755 lab/prove.sh create mode 100644 src/Makefile create mode 100644 src/ringserver-feeder.c diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..af5b86c --- /dev/null +++ b/.dockerignore @@ -0,0 +1,9 @@ +.git +.github +README.md +LICENSE +lab +*.md +.vscode +.gitignore +.DS_Store diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..cd3a4ed --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,22 @@ +version: 2 +updates: + - package-ecosystem: "docker" + directory: "/" + schedule: + interval: "weekly" + open-pull-requests-limit: 5 + commit-message: + prefix: "chore" + include: "scope" + ignore: + - dependency-name: "debian" + update-types: ["version-update:semver-major"] + + - package-ecosystem: "github-actions" + directory: "/" + schedule: + interval: "weekly" + open-pull-requests-limit: 5 + commit-message: + prefix: "chore" + include: "scope" diff --git a/.github/workflows/auto-merge.yml b/.github/workflows/auto-merge.yml new file mode 100644 index 0000000..83de167 --- /dev/null +++ b/.github/workflows/auto-merge.yml @@ -0,0 +1,17 @@ +name: Auto-merge Dependabot PRs + +on: + pull_request: + types: [opened, synchronize, reopened, ready_for_review] + check_suite: + types: [completed] + +permissions: + contents: write + pull-requests: write + checks: read + +jobs: + auto-merge: + uses: actionsforge/actions/.github/workflows/dependabot-auto-merge.yml@main + secrets: inherit diff --git a/.github/workflows/build-and-release.yml b/.github/workflows/build-and-release.yml new file mode 100644 index 0000000..35753bc --- /dev/null +++ b/.github/workflows/build-and-release.yml @@ -0,0 +1,17 @@ +name: Build and Release + +on: + push: + branches: + - main + tags: + - "v*" + workflow_dispatch: + +jobs: + build: + permissions: + contents: write + packages: write + uses: actionsforge/actions/.github/workflows/docker-image-release.yml@main + secrets: inherit diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..7a1dabc --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,14 @@ +name: CI + +on: + pull_request: + branches: + - main + workflow_dispatch: + +jobs: + build: + uses: actionsforge/actions/.github/workflows/docker-image-validate.yml@main + with: + image-tag: "ringserver-feeder:test" + scan-enabled: false diff --git a/.github/workflows/commitmsg-conform.yml b/.github/workflows/commitmsg-conform.yml new file mode 100644 index 0000000..4940385 --- /dev/null +++ b/.github/workflows/commitmsg-conform.yml @@ -0,0 +1,14 @@ +name: Commit Message Conformance + +on: + pull_request: {} + +permissions: + statuses: write + checks: write + contents: read + pull-requests: read + +jobs: + commitmsg-conform: + uses: actionsforge/actions/.github/workflows/commitmsg-conform.yml@main diff --git a/.github/workflows/markdown-lint.yml b/.github/workflows/markdown-lint.yml new file mode 100644 index 0000000..034b809 --- /dev/null +++ b/.github/workflows/markdown-lint.yml @@ -0,0 +1,14 @@ +name: Markdown Lint + +on: + pull_request: {} + +permissions: + statuses: write + checks: write + contents: read + pull-requests: read + +jobs: + markdown-lint: + uses: actionsforge/actions/.github/workflows/markdown-lint.yml@main diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..f843368 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,60 @@ +# Debian bookworm + EarthScope libs. Pin upstream tags via build args. + +ARG BASE=debian:bookworm-slim +ARG SLINK2DALI_VERSION=v0.8 +ARG LIBDALI_VERSION=develop + +FROM ${BASE} AS buildenv +ARG SLINK2DALI_VERSION +ARG LIBDALI_VERSION +ARG DEBIAN_FRONTEND=noninteractive + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + ca-certificates \ + clang \ + git \ + make \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /build +RUN git clone --depth 1 --branch "${SLINK2DALI_VERSION}" \ + https://github.com/EarthScope/slink2dali.git upstream \ + && rm -rf upstream/libdali \ + && git clone --depth 1 --branch "${LIBDALI_VERSION}" \ + https://github.com/EarthScope/libdali.git upstream/libdali + +COPY src/ringserver-feeder.c upstream/src/ +COPY src/Makefile upstream/src/ + +RUN cd upstream \ + && make CC=clang \ + && make -C src CC=clang + +FROM ${BASE} +ARG DEBIAN_FRONTEND=noninteractive +ARG UID=10000 +ARG GID=10000 +ARG USERNAME=containeruser + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + netbase \ + procps \ + && rm -rf /var/cache/apt/archives /var/lib/apt/lists/* + +COPY --from=buildenv /build/upstream/ringserver-feeder /ringserver-feeder + +RUN groupadd --gid "${GID}" "${USERNAME}" \ + && adduser --uid "${UID}" --gid "${GID}" "${USERNAME}" \ + && mkdir -p /data \ + && chown -R "${UID}:${GID}" /data + +WORKDIR /data +USER ${USERNAME} + +ENV FEEDER_STATE_FILE=/data/seedlink.state +ENV FEEDER_PKTID_FILE=/data/pktid.state +ENV FEEDER_LOCK_FILE=/data/feeder.lock + +ENTRYPOINT ["/ringserver-feeder"] diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..48e18e5 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Platform Fuzz + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 3fcf643..ccaabba 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,51 @@ # ringserver-feeder -SeedLink to DataLink feeder with shared packet IDs + +![CI](https://github.com/platformfuzz/ringserver-feeder/actions/workflows/ci.yml/badge.svg) +![Build and Release](https://github.com/platformfuzz/ringserver-feeder/actions/workflows/build-and-release.yml/badge.svg) + +SeedLink client that writes the same DataLink packet ID to every ringserver replica. + +Pulls upstream SeedLink and fans each miniSEED record out with DataLink 1.1 +`WRITE` flag `I` so NLB clients can resume on any replica. Derived from EarthScope +slink2dali; uses libdali `dl_write_id()`. + +**Package:** [ghcr.io/platformfuzz/ringserver-feeder](https://github.com/platformfuzz/ringserver-feeder/pkgs/container/ringserver-feeder) + +## Run + +```bash +docker pull ghcr.io/platformfuzz/ringserver-feeder:latest +docker run --rm \ + -e FEEDER_SEEDLINK_HOST=upstream:18000 \ + -e FEEDER_DATALINK_HOSTS=rs0:16000,rs1:16000 \ + -v feeder-data:/data \ + ghcr.io/platformfuzz/ringserver-feeder:latest +``` + +| Variable | Purpose | +| --- | --- | +| `FEEDER_SEEDLINK_HOST` | Upstream SeedLink `host:port` | +| `FEEDER_DATALINK_HOSTS` | Comma-separated DataLink targets | +| `FEEDER_STATE_FILE` | SeedLink resume state (default `/data/seedlink.state`) | +| `FEEDER_PKTID_FILE` | Last written packet ID (default `/data/pktid.state`) | +| `FEEDER_LOCK_FILE` | Single-writer lock (default `/data/feeder.lock`) | + +slink2dali-style flags (`-s`, `-S`, `-l`, `-x`, `-nd`, `-nt`, `-k`) also work. + +## Lab + +```bash +cd lab +docker compose up --build -d +./prove.sh +docker compose down +``` + +Upstream ringserver scans `lab/miniseed/`. The feeder writes to `rs0` and `rs1`. +`prove.sh` checks both replicas expose the same latest DataLink packet ID. + +## Build + +```bash +docker build -t ringserver-feeder:test . +``` diff --git a/lab/config/replica.conf b/lab/config/replica.conf new file mode 100644 index 0000000..c6e61cb --- /dev/null +++ b/lab/config/replica.conf @@ -0,0 +1,3 @@ +RingDirectory /data/ring +RingSize 32m +WriteIP 0.0.0.0/0 diff --git a/lab/config/upstream.conf b/lab/config/upstream.conf new file mode 100644 index 0000000..d7201ab --- /dev/null +++ b/lab/config/upstream.conf @@ -0,0 +1,3 @@ +RingDirectory /data/ring +RingSize 32m +MSeedScan /seed Match=.* Reject= InitCurrentState=y diff --git a/lab/docker-compose.yml b/lab/docker-compose.yml new file mode 100644 index 0000000..c84355f --- /dev/null +++ b/lab/docker-compose.yml @@ -0,0 +1,45 @@ +services: + upstream: + image: earthscope/ringserver:4.5.6 + volumes: + - ./miniseed:/seed:ro + - ./config/upstream.conf:/config/ringserver.conf:ro + - upstream-data:/data + environment: + RS_CONFIG_FILE: /config/ringserver.conf + + rs0: + image: earthscope/ringserver:4.5.6 + volumes: + - ./config/replica.conf:/config/ringserver.conf:ro + - rs0-data:/data + environment: + RS_CONFIG_FILE: /config/ringserver.conf + + rs1: + image: earthscope/ringserver:4.5.6 + volumes: + - ./config/replica.conf:/config/ringserver.conf:ro + - rs1-data:/data + environment: + RS_CONFIG_FILE: /config/ringserver.conf + + feeder: + build: .. + restart: on-failure + depends_on: + - upstream + - rs0 + - rs1 + volumes: + - feeder-data:/data + environment: + FEEDER_SEEDLINK_HOST: upstream:18000 + FEEDER_DATALINK_HOSTS: rs0:16000,rs1:16000 + FEEDER_VERBOSE: "1" + +volumes: + upstream-data: + rs0-data: + rs1-data: + feeder-data: diff --git a/lab/miniseed/test.mseed b/lab/miniseed/test.mseed new file mode 100644 index 0000000000000000000000000000000000000000..0824c0b33fc03d9b3c121fe71a7f2edabb6d68ef GIT binary patch literal 4096 zcmeH{`&Z0)7{|ZgnWoTXrKHv^id3Q_t&sK1Op!!FgmiQ5;u5>*HuKC3QWCOjn-!FMNXl$d%a~R(^WB;8SL`|Sob$Pz^SsaNb6!6@?(Vu;ZNJuc z&02eVdzRnEkPwqj;xM5v@)I$GAhlQBA_ySd>AeJDN}B1Px{Xdz!Man{p=9y4nq5ku zO)+K{OBb)am2c}&8!^h#uV(JdVtWg_=+Uzr1T{7e(`||F_9tDW{oHv|sTY#5F-1k< z=QQ_#KzDV+`GR`Jl0ETt;XRj?SGum&WD5(MrerFH0+fq}O1yJguZxaqT3ovOv*PSp z+Jl3d1G%4Fii(QfZ78?7chmBt%xBYIcfC2qVj1@)&w(%H@1Rxi>Zpt2*Qz{mn8uAa zUr-?!N$=&`;BN7z6uWAXI#!qGh0;}mPmDZ%I%Qj(lh}QyKo!c%KK$&^(h*hy7w)M$ ziHhsDos&Cx-RYTW9;RmwT5u#)PY10l%@XhG^86iUg7o3h#fMg^_!Yb5QMc%d27%j{ zRavxg18J7Yr*C|SC@B>U<{sR9=E14wQ^V7b9youk`LU>Y=imDL;n_~JbNxy~j**|7 zk!m}KlsKd?eoOk1e=1i35vbL^&qE{{~Hw#REB z5M^CJeAMLrvX0 z)P#GWMkqyU`x+p4uYlwTfm8$ldDR$*n?j@_|3uB4_o%*LhibLsH@-6mhhXz&z}^24Zpy8^#%FUO(R8vHsoK+jiKD}i=;JQ&wH z!iW=HU}AkAXd)@9o+Y8OeIY7tl%Xi%`dZ8Tiu)m;0&;LsQ3L`F& zV6AKcyAB~(H_QVQ;|5g3+`xC&g79^JJ@z;bpi&Y7W*_=s(g`U{ThD`uw-@XABa1XR z;Q13u$4|f~ll-uA?|S@P+z3`n6)-cy89WCa!R3eo%zaF;H=z}qezeECu5YnrYM-7z zkQ@yYDr5j}_^_f^0^Uy;FlJgVb}D9JX^B4;hIeDR4~6oUFmMrn4`G5X*vo$p>%NTC z^KGja;Dgr}F>5dz6I&mmaD)YRIX#BOnfD>1@D&usS3v4=bC@Em!m`Xr+#nx7??-zu zHddhLcf9w4ur0qp<2Dh<1tgUI9u020oH2_Six!@8?mMPAx>mL0gC)Lj%(|QDP#(iv zof-n_k|I5S(L^BzZx?d&N9S^*H?z5IPusAP)d>$P*E2;Yy_mhJSC}zAFCeJK6zP*~ zIaJ3n&P~q}ZlGhYo?meF1Jk1^loj3hfmPEnz+yO^fw*&GZbtkjj!k=QYH5Uo^P8Ir z54wag8!B_yg4%xe&?|4&8-|^pf5JJM(|#@|rOr4jWkKom)Z>ZcF!YDptioj*eaA0b z>RZ*=>N7zqg{n7_)Mlf#Nzao;CrwK1OR>p2tLM8$vDj}Z-=(|kTm9J4GkjliF2JCz zR|{b8rXEEE1TlUs@= 2 and parts[0] in ("OK", "ERROR"): + print(parts[1]) +else: + print("0") +PY +} + +id0="$(read_pktid rs0)" +id1="$(read_pktid rs1)" + +echo "rs0 latest pktid: $id0" +echo "rs1 latest pktid: $id1" + +if [[ -n "$id0" && -n "$id1" && "$id0" == "$id1" && "$id0" != "0" ]]; then + echo "PASS: replicas share packet IDs" + exit 0 +fi + +echo "FAIL: packet IDs differ or ring is empty" +exit 1 diff --git a/src/Makefile b/src/Makefile new file mode 100644 index 0000000..721367b --- /dev/null +++ b/src/Makefile @@ -0,0 +1,18 @@ +# Build ringserver-feeder against EarthScope libs cloned in the Docker build stage. + +CFLAGS ?= -O3 +CFLAGS += -I../libslink -I../libdali -I../libmseed + +LDFLAGS = -L../libslink -L../libdali -L../libmseed +LDLIBS = -lslink -ldali -lmseed + +BIN = ../ringserver-feeder +OBJS = ringserver-feeder.o + +all: $(BIN) + +$(BIN): $(OBJS) + $(CC) $(CFLAGS) -o $(BIN) $(OBJS) $(LDFLAGS) $(LDLIBS) + +clean: + rm -f $(OBJS) $(BIN) diff --git a/src/ringserver-feeder.c b/src/ringserver-feeder.c new file mode 100644 index 0000000..166c34b --- /dev/null +++ b/src/ringserver-feeder.c @@ -0,0 +1,653 @@ +/*************************************************************************** + * ringserver-feeder.c + * + * SeedLink client that forwards miniSEED to multiple DataLink servers using + * the same packet ID on each (DataLink 1.1 WRITE flag I). + * + * Derived from EarthScope slink2dali (Apache-2.0). + ***************************************************************************/ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#define PACKAGE "ringserver-feeder" +#define VERSION "0.1.0" + +typedef struct DLTarget_s +{ + char *address; + DLCP *conn; +} DLTarget; + +static short int verbose = 0; +static int stateint = 0; +static char *netcode = 0; +static char *statefile = 0; +static char *pktidfile = 0; +static char *lockfile = 0; +static int writeack = 0; +static int dialup = 0; +static int keepalive = 300; +static int netto = 600; +static int netdly = 30; + +static SLCD *slconn; +static DLTarget *targets = NULL; +static int targetcount = 0; +static int lockfd = -1; +static uint64_t nextpktid = 1; + +static int sendrecord (char *record, int reclen); +static int connect_targets (void); +static int parameter_proc (int argcount, char **argvec); +static char *getoptval (int argcount, char **argvec, int argopt); +static void term_handler (int sig); +static void print_timelogc (const char *msg); +static void print_timelog (char *msg); +static void usage (void); +static void apply_env (void); +static int parse_hosts (const char *csv); +static int load_pktid (void); +static int save_pktid (uint64_t pktid); +static int acquire_lock (const char *path); + +int +main (int argc, char **argv) +{ + SLpacket *slpack; + int seqnum; + int ptype; + int packetcnt = 0; + + char *type[] = {"Data", "Detection", "Calibration", "Timing", + "Message", "General", "Request", "Info", + "Info (terminated)", "KeepAlive"}; + + struct sigaction sa; + + sa.sa_flags = SA_RESTART; + sigemptyset (&sa.sa_mask); + sa.sa_handler = term_handler; + sigaction (SIGINT, &sa, NULL); + sigaction (SIGQUIT, &sa, NULL); + sigaction (SIGTERM, &sa, NULL); + sa.sa_handler = SIG_IGN; + sigaction (SIGHUP, &sa, NULL); + sigaction (SIGPIPE, &sa, NULL); + + apply_env (); + + if (parameter_proc (argc, argv) < 0) + { + fprintf (stderr, "Argument processing failed\n"); + fprintf (stderr, "Try '-h' for detailed help\n"); + return -1; + } + + if (lockfile && acquire_lock (lockfile) < 0) + { + sl_log (2, 0, "Another feeder holds %s\n", lockfile); + return -1; + } + + if (load_pktid () < 0) + return -1; + + if (connect_targets () < 0) + { + sl_log (2, 0, "Error connecting to one or more DataLink targets\n"); + return -1; + } + + while (sl_collect (slconn, &slpack)) + { + ptype = sl_packettype (slpack); + seqnum = sl_sequence (slpack); + + if (verbose > 1) + { + if (ptype == SLKEEP) + sl_log (1, 0, "Keep alive packet received\n"); + else + sl_log (1, 0, "Received %s packet, SeedLink sequence %d\n", + type[ptype], seqnum); + } + + if (ptype >= SLDATA && ptype < SLNUM) + { + while (sendrecord ((char *)&slpack->msrecord, SLRECSIZE)) + { + if (verbose) + sl_log (1, 0, "Re-connecting to DataLink target(s)\n"); + + for (int i = 0; i < targetcount; i++) + { + if (targets[i].conn->link != -1) + dl_disconnect (targets[i].conn); + } + + if (connect_targets () < 0) + { + sl_log (2, 0, "Error re-connecting to DataLink target(s), sleeping 10 seconds\n"); + sleep (10); + } + + if (slconn->terminate) + break; + } + + packetcnt++; + } + + if (statefile && stateint) + { + if (++packetcnt >= stateint) + { + sl_savestate (slconn, statefile); + packetcnt = 0; + } + } + } + + if (slconn->link != -1) + sl_disconnect (slconn); + + for (int i = 0; i < targetcount; i++) + { + if (targets[i].conn->link != -1) + dl_disconnect (targets[i].conn); + } + + if (statefile) + sl_savestate (slconn, statefile); + + if (lockfd >= 0) + { + flock (lockfd, LOCK_UN); + close (lockfd); + } + + return 0; +} + +static int +connect_targets (void) +{ + for (int i = 0; i < targetcount; i++) + { + if (targets[i].conn->link >= 0) + continue; + + if (dl_connect (targets[i].conn) < 0) + { + sl_log (2, 0, "Error connecting to DataLink server %s\n", targets[i].address); + return -1; + } + } + + return 0; +} + +static int +sendrecord (char *record, int reclen) +{ + static MSRecord *msr = NULL; + hptime_t endtime; + char streamid[100]; + int rv; + uint64_t pktid; + + if (netcode) + ms_strncpopen (((struct fsdh_s *)record)->network, netcode, 2); + + if ((rv = msr_unpack (record, reclen, &msr, 0, 0)) != MS_NOERROR) + { + ms_recsrcname (record, streamid, 0); + sl_log (2, 0, "Error unpacking %s: %s", streamid, ms_errorstr (rv)); + return -1; + } + + msr_srcname (msr, streamid, 0); + strcat (streamid, "/MSEED"); + endtime = msr_endtime (msr); + + if (nextpktid > LIBDALI_PKTID_MAXIMUM) + { + sl_log (2, 0, "Packet ID overflow\n"); + return -1; + } + + pktid = nextpktid++; + + for (int i = 0; i < targetcount; i++) + { + DLTarget *target = &targets[i]; + + if (target->conn->link < 0 && connect_targets () < 0) + return -1; + + if (dl_write_id (target->conn, record, (size_t)reclen, streamid, + msr->starttime, endtime, pktid, writeack) < 0) + { + dl_disconnect (target->conn); + return -1; + } + } + + if (save_pktid (pktid) < 0) + sl_log (2, 0, "Warning: failed to save packet ID state\n"); + + return 0; +} + +static void +apply_env (void) +{ + const char *value; + + if ((value = getenv ("FEEDER_VERBOSE")) && value[0]) + verbose = (short int)atoi (value); + + if ((value = getenv ("FEEDER_STATE_FILE")) && value[0]) + statefile = strdup (value); + + if ((value = getenv ("FEEDER_PKTID_FILE")) && value[0]) + pktidfile = strdup (value); + + if ((value = getenv ("FEEDER_LOCK_FILE")) && value[0]) + lockfile = strdup (value); + + if ((value = getenv ("FEEDER_RECONNECT_SECONDS")) && value[0]) + netdly = atoi (value); + + if ((value = getenv ("FEEDER_NETWORK_TIMEOUT")) && value[0]) + netto = atoi (value); + + if ((value = getenv ("FEEDER_KEEPALIVE_SECONDS")) && value[0]) + keepalive = atoi (value); + + if ((value = getenv ("FEEDER_DIALUP")) && value[0] && strcmp (value, "0") != 0) + dialup = 1; + + if ((value = getenv ("FEEDER_WRITE_ACK")) && value[0] && strcmp (value, "0") != 0) + writeack = 1; +} + +static int +parse_hosts (const char *csv) +{ + char *copy; + char *cursor; + char *token; + int count = 0; + + if (!csv || !csv[0]) + return -1; + + copy = strdup (csv); + if (!copy) + return -1; + + for (token = strtok_r (copy, ",", &cursor); token; token = strtok_r (NULL, ",", &cursor)) + { + while (*token == ' ') + token++; + + if (*token == '\0') + continue; + + targets = realloc (targets, (size_t)(count + 1) * sizeof (DLTarget)); + if (!targets) + { + free (copy); + return -1; + } + + targets[count].address = strdup (token); + targets[count].conn = dl_newdlcp (targets[count].address, PACKAGE); + if (!targets[count].address || !targets[count].conn) + { + free (copy); + return -1; + } + + count++; + } + + free (copy); + targetcount = count; + return (count > 0) ? 0 : -1; +} + +static int +load_pktid (void) +{ + FILE *fp; + uint64_t value; + + if (!pktidfile) + return 0; + + fp = fopen (pktidfile, "r"); + if (!fp) + return 0; + + if (fscanf (fp, "%" SCNu64, &value) == 1 && value > 0) + nextpktid = value + 1; + + fclose (fp); + return 0; +} + +static int +save_pktid (uint64_t pktid) +{ + FILE *fp; + + if (!pktidfile) + return 0; + + fp = fopen (pktidfile, "w"); + if (!fp) + return -1; + + fprintf (fp, "%" PRIu64 "\n", pktid); + fclose (fp); + return 0; +} + +static int +acquire_lock (const char *path) +{ + lockfd = open (path, O_RDWR | O_CREAT, 0644); + if (lockfd < 0) + return -1; + + if (flock (lockfd, LOCK_EX | LOCK_NB) < 0) + { + close (lockfd); + lockfd = -1; + return -1; + } + + return 0; +} + +static int +parameter_proc (int argcount, char **argvec) +{ + const char *value; + char *sladdress = 0; + char *dlhosts = 0; + int error = 0; + char *streamfile = 0; + char *multiselect = 0; + char *selectors = 0; + char *timewin = 0; + char *tptr; + SLstrlist *timelist; + + if ((value = getenv ("FEEDER_SEEDLINK_HOST")) && value[0] && !sladdress) + sladdress = strdup (value); + + if ((value = getenv ("FEEDER_DATALINK_HOSTS")) && value[0] && !dlhosts) + dlhosts = strdup (value); + + if ((value = getenv ("FEEDER_SELECTORS")) && value[0] && !selectors) + selectors = strdup (value); + + if ((value = getenv ("FEEDER_STREAMS")) && value[0] && !multiselect) + multiselect = strdup (value); + + if (!pktidfile && (value = getenv ("FEEDER_PKTID_FILE")) && value[0]) + pktidfile = strdup (value); + + if (!lockfile && (value = getenv ("FEEDER_LOCK_FILE")) && value[0]) + lockfile = strdup (value); + + if (!statefile && (value = getenv ("FEEDER_STATE_FILE")) && value[0]) + statefile = strdup (value); + + if (!pktidfile) + pktidfile = strdup ("/data/pktid.state"); + + if (!lockfile) + lockfile = strdup ("/data/feeder.lock"); + + for (optind = 1; optind < argcount; optind++) + { + if (strcmp (argvec[optind], "-V") == 0) + { + fprintf (stderr, "%s version: %s\n", PACKAGE, VERSION); + exit (0); + } + else if (strcmp (argvec[optind], "-h") == 0) + { + usage (); + exit (0); + } + else if (strncmp (argvec[optind], "-v", 2) == 0) + { + verbose += (short int)strspn (&argvec[optind][1], "v"); + } + else if (strcmp (argvec[optind], "-l") == 0) + streamfile = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-s") == 0) + selectors = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-S") == 0) + multiselect = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-d") == 0) + dialup = 1; + else if (strcmp (argvec[optind], "-N") == 0) + netcode = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-x") == 0) + statefile = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-tw") == 0) + timewin = getoptval (argcount, argvec, optind++); + else if (strcmp (argvec[optind], "-nt") == 0) + netto = atoi (getoptval (argcount, argvec, optind++)); + else if (strcmp (argvec[optind], "-nd") == 0) + netdly = atoi (getoptval (argcount, argvec, optind++)); + else if (strcmp (argvec[optind], "-k") == 0) + keepalive = atoi (getoptval (argcount, argvec, optind++)); + else if (strncmp (argvec[optind], "-", 1) == 0) + { + fprintf (stderr, "Unknown option: %s\n", argvec[optind]); + exit (1); + } + else if (!sladdress) + sladdress = argvec[optind]; + else if (!dlhosts) + dlhosts = argvec[optind]; + else + { + fprintf (stderr, "Unknown option: %s\n", argvec[optind]); + exit (1); + } + } + + if (!sladdress) + { + fprintf (stderr, "No SeedLink server specified (FEEDER_SEEDLINK_HOST)\n"); + error = 1; + } + + if (!dlhosts) + { + fprintf (stderr, "No DataLink targets specified (FEEDER_DATALINK_HOSTS)\n"); + error = 1; + } + + if (error) + { + usage (); + exit (1); + } + + if (!(slconn = sl_newslcd ())) + { + fprintf (stderr, "Cannot allocate SeedLink descriptor\n"); + exit (1); + } + + slconn->sladdr = sladdress; + slconn->netto = netto; + slconn->netdly = netdly; + slconn->keepalive = keepalive; + if (dialup) + slconn->dialup = 1; + + if (parse_hosts (dlhosts) < 0) + { + fprintf (stderr, "Invalid DataLink target list\n"); + exit (1); + } + + sl_loginit (verbose, &print_timelogc, NULL, &print_timelogc, NULL); + dl_loginit (verbose, &print_timelog, "", &print_timelog, ""); + setvbuf (stdout, NULL, _IOLBF, 0); + sl_log (1, 0, "%s version: %s\n", PACKAGE, VERSION); + + if (streamfile) + sl_read_streamlist (slconn, streamfile, selectors); + + if (timewin) + { + if (strchr (timewin, ':') == NULL) + { + sl_log (2, 0, "time window not in begin:[end] format\n"); + return -1; + } + + if (sl_strparse (timewin, ":", &timelist) > 2) + { + sl_log (2, 0, "time window not in begin:[end] format\n"); + sl_strparse (NULL, NULL, &timelist); + return -1; + } + + if (strlen (timelist->element) == 0) + { + sl_log (2, 0, "time window must specify a begin time\n"); + sl_strparse (NULL, NULL, &timelist); + return -1; + } + + slconn->begin_time = strdup (timelist->element); + timelist = timelist->next; + + if (timelist != 0) + { + slconn->end_time = strdup (timelist->element); + if (timelist->next != 0) + { + sl_log (2, 0, "malformed time window specification\n"); + sl_strparse (NULL, NULL, &timelist); + return -1; + } + } + + sl_strparse (NULL, NULL, &timelist); + } + + if (multiselect) + { + if (sl_parse_streamlist (slconn, multiselect, selectors) == -1) + return -1; + } + else if (!streamfile) + sl_setuniparams (slconn, selectors, -1, 0); + + if (statefile) + { + if ((tptr = strchr (statefile, ':')) != NULL) + { + char *tail; + + *tptr++ = '\0'; + stateint = (unsigned int)strtoul (tptr, &tail, 0); + + if (*tail || stateint > 1000000000U) + { + sl_log (2, 0, "state saving interval specified incorrectly\n"); + return -1; + } + } + + if (sl_recoverstate (slconn, statefile) < 0) + sl_log (2, 0, "state recovery failed\n"); + } + + return 0; +} + +static char * +getoptval (int argcount, char **argvec, int argopt) +{ + if (argvec == NULL || argvec[argopt] == NULL) + { + fprintf (stderr, "getoptval(): NULL option requested\n"); + exit (1); + } + + if ((argopt + 1) < argcount && *argvec[argopt + 1] != '-') + return argvec[argopt + 1]; + + fprintf (stderr, "Option %s requires a value\n", argvec[argopt]); + exit (1); +} + +static void +term_handler (int sig) +{ + (void)sig; + sl_terminate (slconn); +} + +static void +print_timelogc (const char *msg) +{ + char timestr[100]; + time_t loc_time; + + time (&loc_time); + strcpy (timestr, asctime (localtime (&loc_time))); + timestr[strlen (timestr) - 1] = '\0'; + fprintf (stdout, "%s - %s", timestr, msg); +} + +static void +print_timelog (char *msg) +{ + print_timelogc (msg); +} + +static void +usage (void) +{ + fprintf (stderr, "%s version %s\n\n", PACKAGE, VERSION); + fprintf (stderr, + "Usage: %s [options]\n\n" + "Environment:\n" + " FEEDER_SEEDLINK_HOST upstream SeedLink host:port\n" + " FEEDER_DATALINK_HOSTS comma-separated DataLink host:port list\n" + " FEEDER_STATE_FILE SeedLink resume state (slink2dali -x)\n" + " FEEDER_PKTID_FILE last written packet ID (default /data/pktid.state)\n" + " FEEDER_LOCK_FILE single-writer lock (default /data/feeder.lock)\n" + " FEEDER_SELECTORS default SeedLink selectors (-s)\n" + " FEEDER_STREAMS multi-station list (-S)\n\n" + "Options match slink2dali where applicable (-l, -s, -S, -x, -nd, -nt, -k).\n\n", + PACKAGE); +}