fix: exit properly on context cancel

The previous assumption was to exit when the fd is closed, but exiting
on context cancelation also makes sense.

Rekres, DRY the code, use generic versions of channel send with context.

Add check for goroutine leak to the tests.

Fixes #4

Signed-off-by: Andrey Smirnov <andrey.smirnov@talos-systems.com>

.dockerignore

@@ -1,8 +1,8 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2021-05-04T14:58:40Z by kres 1844a99-dirty.
+# Generated on 2023-03-02T13:41:04Z by kres latest.
 
-**
+*
 !kmsg.go
 !message.go
 !message_test.go

.drone.yml

@@ -1,7 +1,7 @@
 ---
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2022-10-31T18:04:22Z by kres 03328da.
+# Generated on 2023-03-02T13:41:04Z by kres latest.
 
 kind: pipeline
 type: kubernetes
@@ -174,7 +174,7 @@ steps:
 
 services:
 - name: docker
-  image: docker:20.10-dind
+  image: docker:23.0-dind
   entrypoint:
   - dockerd
   commands:
@@ -212,6 +212,10 @@ trigger:
     exclude:
     - renovate/*
     - dependabot/*
+  event:
+    exclude:
+    - promote
+    - cron
 
 ---
 kind: pipeline

.golangci.yml

@@ -1,6 +1,6 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2022-10-31T18:04:22Z by kres 03328da.
+# Generated on 2023-03-02T13:41:04Z by kres latest.
 
 # options for analysis running
 run:
@@ -150,6 +150,12 @@ linters:
     - maligned
     - golint
     - scopelint
+    - varcheck
+    - deadcode
+    - structcheck
+    - ifshort
+    # disabled as it seems to be broken - goes into imported libraries and reports issues there
+    - musttag
 
 issues:
   exclude: []

Dockerfile

@@ -1,18 +1,18 @@
-# syntax = docker/dockerfile-upstream:1.2.0-labs
+# syntax = docker/dockerfile-upstream:1.5.2-labs
 
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2022-10-31T18:04:22Z by kres 03328da.
+# Generated on 2023-03-02T13:41:04Z by kres latest.
 
 ARG TOOLCHAIN
 
 # cleaned up specs and compiled versions
 FROM scratch AS generate
 
 # runs markdownlint
-FROM docker.io/node:19.0.0-alpine3.16 AS lint-markdown
+FROM docker.io/node:19.7.0-alpine3.16 AS lint-markdown
 WORKDIR /src
-RUN npm i -g markdownlint-cli@0.32.2
+RUN npm i -g markdownlint-cli@0.33.0
 RUN npm i sentences-per-line@0.2.1
 COPY .markdownlint.json .
 COPY ./README.md ./README.md
@@ -29,18 +29,18 @@ ARG CGO_ENABLED
 ENV CGO_ENABLED ${CGO_ENABLED}
 ENV GOPATH /go
 ARG GOLANGCILINT_VERSION
-RUN go install github.com/golangci/golangci-lint/cmd/golangci-lint@${GOLANGCILINT_VERSION} \
+RUN --mount=type=cache,target=/root/.cache/go-build --mount=type=cache,target=/go/pkg go install github.com/golangci/golangci-lint/cmd/golangci-lint@${GOLANGCILINT_VERSION} \
 	&& mv /go/bin/golangci-lint /bin/golangci-lint
 ARG GOFUMPT_VERSION
 RUN go install mvdan.cc/gofumpt@${GOFUMPT_VERSION} \
 	&& mv /go/bin/gofumpt /bin/gofumpt
-RUN go install golang.org/x/vuln/cmd/govulncheck@latest \
+RUN --mount=type=cache,target=/root/.cache/go-build --mount=type=cache,target=/go/pkg go install golang.org/x/vuln/cmd/govulncheck@latest \
 	&& mv /go/bin/govulncheck /bin/govulncheck
 ARG GOIMPORTS_VERSION
-RUN go install golang.org/x/tools/cmd/goimports@${GOIMPORTS_VERSION} \
+RUN --mount=type=cache,target=/root/.cache/go-build --mount=type=cache,target=/go/pkg go install golang.org/x/tools/cmd/goimports@${GOIMPORTS_VERSION} \
 	&& mv /go/bin/goimports /bin/goimports
 ARG DEEPCOPY_VERSION
-RUN go install github.com/siderolabs/deep-copy@${DEEPCOPY_VERSION} \
+RUN --mount=type=cache,target=/root/.cache/go-build --mount=type=cache,target=/go/pkg go install github.com/siderolabs/deep-copy@${DEEPCOPY_VERSION} \
 	&& mv /go/bin/deep-copy /bin/deep-copy
 
 # tools and sources

Makefile

@@ -1,6 +1,6 @@
 # THIS FILE WAS AUTOMATICALLY GENERATED, PLEASE DO NOT EDIT.
 #
-# Generated on 2022-10-31T18:04:22Z by kres 03328da.
+# Generated on 2023-03-02T13:41:04Z by kres latest.
 
 # common variables
 
@@ -13,14 +13,14 @@ WITH_RACE ?= false
 REGISTRY ?= ghcr.io
 USERNAME ?= siderolabs
 REGISTRY_AND_USERNAME ?= $(REGISTRY)/$(USERNAME)
-GOLANGCILINT_VERSION ?= v1.50.1
+GOLANGCILINT_VERSION ?= v1.51.2
 GOFUMPT_VERSION ?= v0.4.0
-GO_VERSION ?= 1.19
-GOIMPORTS_VERSION ?= v0.2.0
+GO_VERSION ?= 1.20
+GOIMPORTS_VERSION ?= v0.6.0
 PROTOBUF_GO_VERSION ?= 1.28.1
 GRPC_GO_VERSION ?= 1.2.0
-GRPC_GATEWAY_VERSION ?= 2.12.0
-VTPROTOBUF_VERSION ?= 0.3.0
+GRPC_GATEWAY_VERSION ?= 2.15.1
+VTPROTOBUF_VERSION ?= 0.4.0
 DEEPCOPY_VERSION ?= v0.5.5
 GO_BUILDFLAGS ?=
 GO_LDFLAGS ?=
@@ -58,7 +58,7 @@ COMMON_ARGS += --build-arg=GRPC_GATEWAY_VERSION="$(GRPC_GATEWAY_VERSION)"
 COMMON_ARGS += --build-arg=VTPROTOBUF_VERSION="$(VTPROTOBUF_VERSION)"
 COMMON_ARGS += --build-arg=DEEPCOPY_VERSION="$(DEEPCOPY_VERSION)"
 COMMON_ARGS += --build-arg=TESTPKGS="$(TESTPKGS)"
-TOOLCHAIN ?= docker.io/golang:1.19-alpine
+TOOLCHAIN ?= docker.io/golang:1.20-alpine
 
 # help menu
 

go.mod

@@ -1,14 +1,17 @@
 module github.com/siderolabs/go-kmsg
 
-go 1.19
+go 1.20
 
 require (
-	github.com/stretchr/testify v1.8.1
-	golang.org/x/sys v0.1.0
+	github.com/siderolabs/gen v0.4.3
+	github.com/stretchr/testify v1.8.2
+	go.uber.org/goleak v1.2.1
+	golang.org/x/sys v0.5.0
 )
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,19 +1,27 @@
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/siderolabs/gen v0.4.3 h1:V3UsZ2KrsryaTMZGZUHAr1CFdPc2/R1lM6lA4a4zCDo=
+github.com/siderolabs/gen v0.4.3/go.mod h1:wS8tFq7sn5vqKAuyS30vJUig3tX5v6q79VG4KfUnILM=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
-golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

reader.go

@@ -12,6 +12,8 @@ import (
 	"os"
 	"syscall"
 	"time"
+
+	"github.com/siderolabs/gen/channel"
 )
 
 // Packet combines Message and error.
@@ -77,7 +79,7 @@ func NewReader(options ...Option) (Reader, error) {
 	}
 
 	if r.options.tail {
-		_, err = r.f.Seek(0, os.SEEK_END)
+		_, err = r.f.Seek(0, io.SeekEnd)
 		if err != nil {
 			r.f.Close() //nolint:errcheck
 
@@ -116,12 +118,9 @@ func (r *reader) scanNoFollow(ctx context.Context, ch chan<- Packet) {
 	fd := int(r.f.Fd())
 
 	if err := syscall.SetNonblock(fd, true); err != nil {
-		select {
-		case ch <- Packet{
+		channel.SendWithContext(ctx, ch, Packet{
 			Err: fmt.Errorf("error switching to nonblock mode: %w", err),
-		}:
-		case <-ctx.Done():
-		}
+		})
 
 		return
 	}
@@ -141,22 +140,18 @@ func (r *reader) scanNoFollow(ctx context.Context, ch chan<- Packet) {
 				continue
 			}
 
-			select {
-			case ch <- Packet{
+			channel.SendWithContext(ctx, ch, Packet{
 				Err: fmt.Errorf("error reading from kmsg: %w", err),
-			}:
-			case <-ctx.Done():
-			}
+			})
 
 			return
 		}
 
 		var packet Packet
 		packet.Message, packet.Err = ParseMessage(string(buf[:n]), r.bootTime)
 
-		select {
-		case ch <- packet:
-		case <-ctx.Done():
+		if !channel.SendWithContext(ctx, ch, packet) {
+			return
 		}
 	}
 }
@@ -179,22 +174,18 @@ func (r *reader) scanFollow(ctx context.Context, ch chan<- Packet) {
 				continue
 			}
 
-			select {
-			case ch <- Packet{
+			channel.SendWithContext(ctx, ch, Packet{
 				Err: fmt.Errorf("error reading from kmsg: %w", err),
-			}:
-			case <-ctx.Done():
-			}
+			})
 
 			return
 		}
 
 		var packet Packet
 		packet.Message, packet.Err = ParseMessage(string(buf[:n]), r.bootTime)
 
-		select {
-		case ch <- packet:
-		case <-ctx.Done():
+		if !channel.SendWithContext(ctx, ch, packet) {
+			return
 		}
 	}
 }

reader_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/goleak"
 
 	"github.com/siderolabs/go-kmsg"
 )
@@ -29,6 +30,8 @@ func skipIfNoKmsg(t *testing.T) {
 func TestReaderNoFollow(t *testing.T) {
 	skipIfNoKmsg(t)
 
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
 	r, err := kmsg.NewReader()
 	assert.NoError(t, err)
 
@@ -59,6 +62,8 @@ func TestReaderFollowTail(t *testing.T) {
 func testReaderFollow(t *testing.T, expectMessages bool, options ...kmsg.Option) {
 	skipIfNoKmsg(t)
 
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
 	r, err := kmsg.NewReader(append([]kmsg.Option{kmsg.Follow()}, options...)...)
 	assert.NoError(t, err)