BE-594 | Delivery package

delivery/grpc/grpc.go

@@ -0,0 +1,68 @@
+// Package grpc provides a custom gRPC client implementation for Cosmos SDK-based applications.
+package grpc
+
+import (
+	"fmt"
+
+	proto "github.com/cosmos/gogoproto/proto"
+
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/encoding"
+)
+
+type OsmomathCodec struct {
+	parentCodec encoding.Codec
+}
+
+func (c OsmomathCodec) Marshal(v interface{}) ([]byte, error) {
+	protoMsg, ok := v.(proto.Message)
+	if !ok {
+		return nil, fmt.Errorf("failed to assert proto.Message")
+	}
+	return proto.Marshal(protoMsg)
+}
+
+func (c OsmomathCodec) Unmarshal(data []byte, v interface{}) error {
+	protoMsg, ok := v.(proto.Message)
+	if !ok {
+		return fmt.Errorf("failed to assert proto.Message")
+	}
+	return proto.Unmarshal(data, protoMsg)
+}
+
+func (c OsmomathCodec) Name() string {
+	return "gogoproto"
+}
+
+// Client wraps a gRPC ClientConn, providing a custom connection.
+// Connection is set up with custom options, including the use of a custom codec
+// for gogoproto and OpenTelemetry instrumentation.
+// Client addresses marshaling math.LegacyDec issue: https://github.com/cosmos/cosmos-sdk/issues/18430
+type Client struct {
+	*grpc.ClientConn
+}
+
+// NewClient creates a new gRPC client connection to the specified endpoint.
+func NewClient(grpcEndpoint string) (*Client, error) {
+	customCodec := &OsmomathCodec{parentCodec: encoding.GetCodec("proto")}
+
+	grpcOpts := []grpc.DialOption{
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
+		grpc.WithDefaultCallOptions(grpc.ForceCodec(customCodec)),
+	}
+
+	grpcConn, err := grpc.NewClient(
+		grpcEndpoint,
+		grpcOpts...,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial Cosmos gRPC service: %w", err)
+	}
+
+	return &Client{
+		ClientConn: grpcConn,
+	}, nil
+}

delivery/grpc/grpc_test.go

@@ -0,0 +1,38 @@
+package grpc_test
+
+import (
+	"testing"
+
+	"github.com/osmosis-labs/sqs/delivery/grpc"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// TestNewClient tests the NewClient function
+func TestNewClient(t *testing.T) {
+	tests := []struct {
+		name     string
+		endpoint string
+		wantErr  bool
+	}{
+		{
+			name:     "Valid endpoint",
+			endpoint: "localhost:9090",
+			wantErr:  false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			client, err := grpc.NewClient(tt.endpoint)
+			if tt.wantErr {
+				assert.Error(t, err)
+				assert.Nil(t, client)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, client)
+				assert.NotNil(t, client.ClientConn)
+			}
+		})
+	}
+}

delivery/http/get.go

@@ -0,0 +1,38 @@
+package http
+
+import (
+	"context"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"time"
+)
+
+// Get issues GET request to given URL using default httpClient.
+func Get(ctx context.Context, url string) ([]byte, error) {
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := DefaultClient.Do(req)
+	if err != nil {
+		netErr, ok := err.(net.Error)
+		if ok && netErr.Timeout() {
+			log.Printf("Request to %s timed out, continuing...", url)
+			return nil, nil
+		}
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	return body, nil
+}

delivery/http/get_test.go

@@ -0,0 +1,71 @@
+package http_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	sqshttp "github.com/osmosis-labs/sqs/delivery/http"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGet(t *testing.T) {
+	tests := []struct {
+		name           string
+		url            string
+		expectedBody   string
+		timeout        time.Duration
+		serverResponse func(w http.ResponseWriter, r *http.Request)
+	}{
+		{
+			name:         "Success",
+			url:          "/success",
+			expectedBody: "Hello, World!",
+			serverResponse: func(w http.ResponseWriter, r *http.Request) {
+				w.Write([]byte("Hello, World!"))
+			},
+		},
+		{
+			name:         "Timeout",
+			url:          "/timeout",
+			expectedBody: "",
+			timeout:      10 * time.Millisecond,
+			serverResponse: func(w http.ResponseWriter, r *http.Request) {
+				time.Sleep(20 * time.Millisecond)
+				w.Write([]byte("Too late"))
+			},
+		},
+		{
+			name:         "Server Error",
+			url:          "/error",
+			expectedBody: "Internal Server Error\n",
+			serverResponse: func(w http.ResponseWriter, r *http.Request) {
+				http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+			},
+		},
+	}
+
+	defaultTimeout := sqshttp.DefaultClient.Timeout
+	resetClient := func() {
+		sqshttp.DefaultClient.Timeout = defaultTimeout
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(tt.serverResponse))
+			defer server.Close()
+
+			sqshttp.DefaultClient.Timeout = tt.timeout
+			defer resetClient()
+
+			ctx := context.Background()
+			body, err := sqshttp.Get(ctx, server.URL+tt.url)
+			assert.NoError(t, err)
+			assert.Equal(t, string(body), tt.expectedBody)
+
+		})
+	}
+}

delivery/http/http.go

@@ -1,10 +1,21 @@
 package http
 
 import (
+	"net/http"
+	"time"
+
 	"github.com/labstack/echo/v4"
 	"github.com/osmosis-labs/sqs/validator"
+
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
+// DefaultClient represents default HTTP client for issuing outgoing HTTP requests.
+var DefaultClient = &http.Client{
+	Timeout:   5 * time.Second, // Adjusted timeout to 5 seconds
+	Transport: otelhttp.NewTransport(http.DefaultTransport),
+}
+
 // RequestUnmarshaler is any type capable to unmarshal data from HTTP request to itself.
 type RequestUnmarshaler interface {
 	UnmarshalHTTPRequest(c echo.Context) error