Connect go v2

generate/go_client.tpl

@@ -3,10 +3,6 @@ package client
 
 import (
 	"context"
-
-	"connectrpc.com/connect"
-	compress "github.com/klauspost/connect-compress/v2"
-
 {{ range $name, $api := . -}}
 	"github.com/metal-stack/api/go{{ $api.Path }}/{{ $api.Name }}connect"
 {{ end }}
@@ -19,11 +15,6 @@ type (
 {{ end }}
 	Ping(context.Context, *PingConfig)
 	}
-	client struct {
-		config *DialConfig
-
-		interceptors []connect.Interceptor
-	}
 {{ range $name, $api := . -}}
 	{{ $name | title }} interface {
 {{ range $svc := $api.Services -}}
@@ -44,12 +35,7 @@ type (
 func (c *client) {{ $name | title }}() {{ $name | title }} {
 	a := &{{ $name }}{
 {{ range $svc := $api.Services -}}
-	{{ $svc.Name | lower }}:  {{ $name }}connect.New{{ $svc.Name }}Client(
-		c.config.HttpClient(),
-		c.config.BaseURL,
-		connect.WithInterceptors(c.interceptors...),
-		compress.WithAll(compress.LevelBalanced),
-	),
+	{{ $svc.Name | lower }}:  {{ $name }}connect.New{{ $svc.Name }}Client(c.httpClient),
 {{ end }}
 	}
 	return a

generate/go_servicepermissions.tpl

@@ -3,7 +3,8 @@ package permissions
 
 import (
 	apiv2 "github.com/metal-stack/api/go/metalstack/api/v2"
-	"connectrpc.com/connect"
+	"connectrpc.com/connect/v2"
+	"google.golang.org/protobuf/proto"
 )
 
 func GetServices() []string {
@@ -113,73 +114,73 @@ func GetServicePermissions() *ServicePermissions {
 	}
 }
 
-func IsPublicScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Public[req.Spec().Procedure]
+func IsPublicScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Public[spec.Procedure]
 	return ok
 }
 
-func IsSelfScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Self[req.Spec().Procedure]
+func IsSelfScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Self[spec.Procedure]
 	return ok
 }
 
-func IsAdminScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Admin[req.Spec().Procedure]
+func IsAdminScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Admin[spec.Procedure]
 	return ok
 }
 
-func IsInfraScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Infra[req.Spec().Procedure]
+func IsInfraScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Infra[spec.Procedure]
 	return ok
 }
 
-func IsMachineScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Machine[req.Spec().Procedure]
+func IsMachineScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Machine[spec.Procedure]
 	return ok
 }
 
-func IsTenantScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Tenant[req.Spec().Procedure]
+func IsTenantScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Tenant[spec.Procedure]
 	return ok
 }
 
-func IsProjectScope(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Visibility.Project[req.Spec().Procedure]
+func IsProjectScope(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Visibility.Project[spec.Procedure]
 	return ok
 }
 
-func IsAuditable(req connect.AnyRequest) bool {
-	_, ok := GetServicePermissions().Auditable[req.Spec().Procedure]
+func IsAuditable(spec connect.Spec) bool {
+	_, ok := GetServicePermissions().Auditable[spec.Procedure]
 	return ok
 }
 
-func GetTenantFromRequest(req connect.AnyRequest) (string, bool) {
-	if !IsTenantScope(req) {
+func GetTenantFromRequest(spec connect.Spec, req proto.Message) (string, bool) {
+	if !IsTenantScope(spec) {
 		return "", false
 	}
-	switch rq := req.Any().(type) {
+	switch rq := req.(type) {
 	case interface{ GetLogin() string }:
 		return rq.GetLogin(), true
 	}
 	return "", false
 }
 
-func GetProjectFromRequest(req connect.AnyRequest) (string, bool) {
-	if !IsProjectScope(req) {
+func GetProjectFromRequest(spec connect.Spec, req proto.Message) (string, bool) {
+	if !IsProjectScope(spec) {
 		return "", false
 	}
-	switch rq := req.Any().(type) {
+	switch rq := req.(type) {
 	case interface{ GetProject() string }:
 		return rq.GetProject(), true
 	}
 	return "", false
 }
 
-func GetMachineIdFromRequest(req connect.AnyRequest) (string, bool) {
-	if !IsMachineScope(req) {
+func GetMachineIdFromRequest(spec connect.Spec, req proto.Message) (string, bool) {
+	if !IsMachineScope(spec) {
 		return "", false
 	}
-	switch rq := req.Any().(type) {
+	switch rq := req.(type) {
 	case interface{ GetUuid() string }:
 		return rq.GetUuid(), true
 	}

go.mod

@@ -5,13 +5,12 @@ go 1.27
 require (
 	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.12-20260825204119-511051f7f437.2
 	buf.build/go/protovalidate v1.4.0
-	connectrpc.com/connect v1.21.0
+	connectrpc.com/connect/v2 v2.0.0
 	github.com/bufbuild/protocompile v0.14.1
 	github.com/go-task/slim-sprig/v3 v3.0.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
-	github.com/klauspost/connect-compress/v2 v2.1.1
 	github.com/stretchr/testify v1.12.1
 	google.golang.org/grpc v1.84.0
 	google.golang.org/protobuf v1.36.12
@@ -21,12 +20,10 @@ require (
 	cel.dev/cel-go v0.32.0 // indirect
 	cel.dev/expr v0.25.3 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
-	github.com/klauspost/compress v1.20.1 // indirect
-	github.com/minio/minlz v1.2.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.5 // indirect
-	golang.org/x/exp v0.0.0-20261007192929-f45ad48fbe92 // indirect
-	golang.org/x/sys v0.48.0 // indirect
-	golang.org/x/text v0.42.0 // indirect
+	golang.org/x/exp v0.0.0-20261009195045-ca0d7ba23607 // indirect
+	golang.org/x/sys v0.49.0 // indirect
+	golang.org/x/text v0.43.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 // indirect
 )

go.sum

@@ -6,8 +6,8 @@ cel.dev/cel-go v0.32.0 h1:irvpFKr5EuGPyxeME03ERh0rii1TX+BDAnB9eL3IvNk=
 cel.dev/cel-go v0.32.0/go.mod h1:DnVip7tpJSsgZymwfT+m1tnEVy3ivAjSMXPx12YrMkU=
 cel.dev/expr v0.25.3 h1:A2jO8jwOugrrovveCWfj0KEZOfqiLgAcwjpHPhzIGw0=
 cel.dev/expr v0.25.3/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
-connectrpc.com/connect v1.21.0 h1:LhqSJt7jHf5NJBo9Jq/t/9FjcYAideif0mg+qe2jCUs=
-connectrpc.com/connect v1.21.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
+connectrpc.com/connect/v2 v2.0.0 h1:PVVpWl2Xv9dIWN2gijs4HK40NSnPFzLqCEEmZ8fCUQ8=
+connectrpc.com/connect/v2 v2.0.0/go.mod h1:N32OLrstcPUzkykFynnefFro3+xzSdOrZlK2jHmS/xo=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
 github.com/brianvoe/gofakeit/v6 v6.28.0 h1:Xib46XXuQfmlLS2EXRuJpqcw8St6qSZz75OUo0tgAW4=
@@ -24,26 +24,20 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
-github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
-github.com/klauspost/connect-compress/v2 v2.1.1 h1:ycZNp4rWOZBodVE2Ls5AzK4aHkyK+GteEfzRZgKNs+c=
-github.com/klauspost/connect-compress/v2 v2.1.1/go.mod h1:9oilsPHJMzGKkjafSBk9J7iVo4mO+dw0G0KSdVpnlVE=
-github.com/minio/minlz v1.2.2 h1:o3YoP6f3VG0oY3Ykhx2mHrpCywsoobsFPVWslgfKcbY=
-github.com/minio/minlz v1.2.2/go.mod h1:xHhZ+MJmC2U4ssK/rSENsoyCfv0VFQo1mT6AdCA2RJ8=
 github.com/rodaine/protogofakeit v0.1.1 h1:ZKouljuRM3A+TArppfBqnH8tGZHOwM/pjvtXe9DaXH8=
 github.com/rodaine/protogofakeit v0.1.1/go.mod h1:pXn/AstBYMaSfc1/RqH3N82pBuxtWgejz1AlYpY1mI0=
 github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
 github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
 go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
 go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
-golang.org/x/exp v0.0.0-20261007192929-f45ad48fbe92 h1:kc+Gi2MFkd0KWtdDlVxVwjw4KB98/77usJDFj+6YFQ8=
-golang.org/x/exp v0.0.0-20261007192929-f45ad48fbe92/go.mod h1:SRjTzy5WcYqRRqquLUvHeeovSwLO0CZ3xcltZNO4Fck=
+golang.org/x/exp v0.0.0-20261009195045-ca0d7ba23607 h1:o259es6R0mYtmCeID0xv+dkp5kQSf6qp8zs0j4MdUaI=
+golang.org/x/exp v0.0.0-20261009195045-ca0d7ba23607/go.mod h1:SRjTzy5WcYqRRqquLUvHeeovSwLO0CZ3xcltZNO4Fck=
 golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
 golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
-golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
-golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
-golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
-golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
+golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
+golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc=
+golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s=
 google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8 h1:cFbkNigDInw+wVTXc+CyhUsoHGJCw0BR6mlNoRVfre0=
 google.golang.org/genproto/googleapis/api v0.0.0-20261005182115-fad411399dd8/go.mod h1:QK6p7IgD7IZEZJBhhSUCib5X2vMOzlrGCGUw2XyVDh0=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 h1:0SXBqli5dnuCKMjI/X3FM226zEaEa0GUcSCJtN1vL2o=

go/client/client-interceptors.go

@@ -9,68 +9,68 @@ import (
 	"sync/atomic"
 	"time"
 
-	"connectrpc.com/connect"
+	"connectrpc.com/connect/v2"
 	apiv2models "github.com/metal-stack/api/go/metalstack/api/v2"
 )
 
-// authinterceptor adds the required auth headers
+// authInterceptor adds the required auth headers
 type authInterceptor struct {
 	config *DialConfig
 }
 
-func (i *authInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
-	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-		request.Header().Add("Authorization", "Bearer "+i.config.Token)
-		return next(ctx, request)
-	})
-}
-
-func (i *authInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
-	return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
-		return &streamingInterceptorConn{
-			StreamingClientConn: next(ctx, spec),
-			token:               i.config.Token,
+func (i *authInterceptor) intercept(next connect.ClientFunc) connect.ClientFunc {
+	return func(ctx context.Context, spec connect.Spec) (connect.ClientStream, error) {
+		if callInfo, ok := connect.CallInfoForClientContext(ctx); ok {
+			callInfo.RequestHeader().Set("Authorization", "Bearer "+i.config.Token)
 		}
+		return next(ctx, spec)
 	}
 }
 
-func (i *authInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
-	return next
+type loggingInterceptor struct {
+	config *DialConfig
 }
 
-type streamingInterceptorConn struct {
-	connect.StreamingClientConn
-	token string
+func (i *loggingInterceptor) intercept(next connect.ClientFunc) connect.ClientFunc {
+	return func(ctx context.Context, spec connect.Spec) (connect.ClientStream, error) {
+		stream, err := next(ctx, spec)
+		if err != nil {
+			return nil, err
+		}
+		return &loggingStream{config: i.config, spec: spec, stream: stream}, nil
+	}
 }
 
-func (conn *streamingInterceptorConn) Send(m any) error {
-	conn.RequestHeader().Add("Authorization", "Bearer "+conn.token)
-	return conn.StreamingClientConn.Send(m)
+type loggingStream struct {
+	config *DialConfig
+	spec   connect.Spec
+	stream connect.ClientStream
 }
 
-type loggingInterceptor struct {
-	config *DialConfig
+func (s *loggingStream) SendHeaders() error {
+	return s.stream.SendHeaders()
 }
 
-func (i *loggingInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
-	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
-		i.config.Log.Debug("intercept", "request procedure", request.Spec().Procedure, "body", request.Any())
-		response, err := next(ctx, request)
-		if err != nil {
-			return nil, err
-		}
-		i.config.Log.Debug("intercept", "request procedure", request.Spec().Procedure, "response", response.Any())
-		return response, err
-	})
+func (s *loggingStream) Send(msg any) error {
+	s.config.Log.Debug("intercept", "request procedure", s.spec.Procedure, "body", msg)
+	return s.stream.Send(msg)
+}
+
+func (s *loggingStream) CloseSend() error {
+	return s.stream.CloseSend()
 }
 
-func (i *loggingInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
-	// TODO also log here
-	return next
+func (s *loggingStream) Receive(msg any) error {
+	err := s.stream.Receive(msg)
+	if err != nil {
+		return err
+	}
+	s.config.Log.Debug("intercept", "request procedure", s.spec.Procedure, "response", msg)
+	return nil
 }
 
-func (i *loggingInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
-	return next
+func (s *loggingStream) Close() error {
+	return s.stream.Close()
 }
 
 type tokenRenewingInterceptor struct {
@@ -82,22 +82,14 @@ type tokenRenewingInterceptor struct {
 	sync.Mutex
 }
 
-func (i *tokenRenewingInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
-	return connect.UnaryFunc(func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
+func (i *tokenRenewingInterceptor) intercept(next connect.ClientFunc) connect.ClientFunc {
+	return func(ctx context.Context, spec connect.Spec) (connect.ClientStream, error) {
 		err := i.renewTokenIfNeeded()
 		if err != nil {
 			return nil, err
 		}
-		return next(ctx, request)
-	})
-}
-
-func (i *tokenRenewingInterceptor) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
-	return next
-}
-
-func (i *tokenRenewingInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
-	return next
+		return next(ctx, spec)
+	}
 }
 
 func (i *tokenRenewingInterceptor) renewTokenIfNeeded() error {