<commit_before>// Package grpcutil implements a gRPC HTTP handler and OpenAPI proxy. Useful to serve gRPC requests from an existing HTTP servers.
package grpcutil

import (
	"crypto/tls"
	"crypto/x509"
	"log"
	"net/http"
	"strings"

	"github.com/grpc-ecosystem/grpc-gateway/runtime"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// ServiceBinding are the gRPC server, and gRPC HTTP gateway to which the service will be bound to.
type ServiceBinding struct {
	GRPCServer        *grpc.Server
	GRPCGatewayClient *grpc.ClientConn
	GRPCGatewayMuxer  *runtime.ServeMux
}

// ServiceRegisterFn defines a function type for registering gRPC services.
type ServiceRegisterFn func(ServiceBinding) error

// Option implements http://commandcenter.blogspot.com/2014/01/self-referential-functions-and-design.html
type Option func(*options)

type options struct {
	serverOpts []grpc.ServerOption
	services   []ServiceRegisterFn
	cert       *tls.Certificate
	port       string
	skipPaths  []string
}

// WithServerOpts sets gRPC server options. Optional.
func WithServerOpts(opts []grpc.ServerOption) Option {
	return func(o *options) {
		o.serverOpts = opts
	}
}

// WithServices sets the list of services to register with the gRPC and OpenAPI servers. Required.
func WithServices(services []ServiceRegisterFn) Option {
	return func(o *options) {
		o.services = services
	}
}

// WithTLSCert sets the TLS certificate to use by the gRPC server and OpenAPI gRPC client. Required.
func WithTLSCert(cert *tls.Certificate) Option {
	return func(o *options) {
		o.cert = cert
	}
}

// WithPort sets the port for the OpenAPI gRPC client to use when connecting to the gRPC service.
func WithPort(port string) Option {
	return func(o *options) {
		o.port = port
	}
}

// WithSkipPath allows other handlers to serve static JSON files by instructing the GRPC Gateway Muxer to
// skip serving the given prefixed paths.
func WithSkipPath(path ...string) Option {
	return func(o *options) {
		o.skipPaths = path
	}
}

// Handler serves gRPC and OpenAPI requests or hands over to the next handler if they are not gRPC
// or application/json requests.
func Handler(h http.Handler, opts ...Option) http.Handler {
	options := new(options)
	for _, opt := range opts {
		opt(options)
	}

	if options.cert == nil {
		log.Fatal("grpcutil: TLS certificate required")
	}

	if options.port == "" {
		log.Fatal("grpcutil: port required")
	}

	if len(options.services) == 0 {
		log.Fatal("grpcutil: the list of gRPC services is required")
	}

	options.serverOpts = append(options.serverOpts, grpc.Creds(credentials.NewServerTLSFromCert(options.cert)))
	server := grpc.NewServer(options.serverOpts...)

	certPool := x509.NewCertPool()
	x509Cert, err := x509.ParseCertificate(options.cert.Certificate[0])
	if err != nil {
		log.Fatalf("grpcutil: failed parsing x509 certificate: %+v, ", err)
	}
	certPool.AddCert(x509Cert)
	clientCreds := credentials.NewClientTLSFromCert(certPool, "")

	clientOpts := []grpc.DialOption{
		grpc.WithTransportCredentials(clientCreds),
		grpc.WithUserAgent("grpc-gw"),
	}

	address := "localhost:" + options.port
	clientConn, err := grpc.Dial(address, clientOpts...)
	if err != nil {
		log.Fatalf("failed to connect to local gRPC server: %v", err)
	}

	gwMuxer := runtime.NewServeMux()

	serviceBinding := ServiceBinding{
		GRPCServer:        server,
		GRPCGatewayClient: clientConn,
		GRPCGatewayMuxer:  gwMuxer,
	}

	log.Println("registering GRPC services...")
	for _, register := range options.services {
		if err := register(serviceBinding); err != nil {
			log.Fatalf("failed to register service: %+v", err)
		}
	}

	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		contentType := r.Header.Get("Content-Type")
		if r.ProtoMajor == 2 && strings.Contains(contentType, "application/grpc") {
			server.ServeHTTP(w, r)
			return
		}

		for _, p := range options.skipPaths {
			if strings.HasPrefix(r.URL.Path, p) {
				h.ServeHTTP(w, r)
				return
			}
		}

		accept := r.Header.Get("Accept")
		if strings.Contains(contentType, "application/json") ||
			strings.Contains(accept, "application/json") {
			gwMuxer.ServeHTTP(w, r)
			return
		}
		h.ServeHTTP(w, r)
	})
}
<commit_msg>grpcutil: Add support for setting compressor and decompressor<commit_after>// Package grpcutil implements a gRPC HTTP handler and OpenAPI proxy. Useful to serve gRPC requests from an existing HTTP servers.
package grpcutil

import (
	"crypto/tls"
	"crypto/x509"
	"log"
	"net/http"
	"strings"

	"github.com/grpc-ecosystem/grpc-gateway/runtime"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// ServiceBinding are the gRPC server, and gRPC HTTP gateway to which the service will be bound to.
type ServiceBinding struct {
	GRPCServer        *grpc.Server
	GRPCGatewayClient *grpc.ClientConn
	GRPCGatewayMuxer  *runtime.ServeMux
}

// ServiceRegisterFn defines a function type for registering gRPC services.
type ServiceRegisterFn func(ServiceBinding) error

// Option implements http://commandcenter.blogspot.com/2014/01/self-referential-functions-and-design.html
type Option func(*options)

type options struct {
	serverOpts   []grpc.ServerOption
	services     []ServiceRegisterFn
	cert         *tls.Certificate
	compressor   grpc.Compressor
	decompressor grpc.Decompressor
	port         string
	skipPaths    []string
}

// WithServerOpts sets gRPC server options. Optional.
func WithServerOpts(opts []grpc.ServerOption) Option {
	return func(o *options) {
		o.serverOpts = opts
	}
}

// WithServices sets the list of services to register with the gRPC and OpenAPI servers. Required.
func WithServices(services []ServiceRegisterFn) Option {
	return func(o *options) {
		o.services = services
	}
}

// WithTLSCert sets the TLS certificate to use by the gRPC server and OpenAPI gRPC client. Required.
func WithTLSCert(cert *tls.Certificate) Option {
	return func(o *options) {
		o.cert = cert
	}
}

// WithPort sets the port for the OpenAPI gRPC client to use when connecting to the gRPC service.
func WithPort(port string) Option {
	return func(o *options) {
		o.port = port
	}
}

// WithCompressor sets gRPC server and gateway client compressor.
// Use this function instead of passing compressor through grcputil.WithServerOpts()
func WithCompressor(c grpc.Compressor) Option {
	return func(o *options) {
		o.compressor = c
	}
}

// WithDecompressor sets gRPC server and gateway client decompressor.
// Use this function instead of passing a decompressor through grcputil.WithServerOpts()
func WithDecompressor(d grpc.Decompressor) Option {
	return func(o *options) {
		o.decompressor = d
	}
}

// WithSkipPath allows other handlers to serve static JSON files by instructing the GRPC Gateway Muxer to
// skip serving the given prefixed paths.
func WithSkipPath(path ...string) Option {
	return func(o *options) {
		o.skipPaths = path
	}
}

// Handler serves gRPC and OpenAPI requests or hands over to the next handler if they are not gRPC
// or application/json requests.
func Handler(h http.Handler, opts ...Option) http.Handler {
	options := new(options)
	for _, opt := range opts {
		opt(options)
	}

	if options.cert == nil {
		log.Fatal("grpcutil: TLS certificate required")
	}

	if options.port == "" {
		log.Fatal("grpcutil: port required")
	}

	if len(options.services) == 0 {
		log.Fatal("grpcutil: the list of gRPC services is required")
	}

	if options.cert == nil {
		log.Fatal("grpcutil: a TLS key pair is required")
	}

	options.serverOpts = append(options.serverOpts,
		grpc.Creds(credentials.NewServerTLSFromCert(options.cert)))

	if options.compressor != nil {
		options.serverOpts = append(options.serverOpts,
			grpc.RPCCompressor(options.compressor))
	}
	if options.decompressor != nil {
		options.serverOpts = append(options.serverOpts,
			grpc.RPCDecompressor(options.decompressor))
	}

	server := grpc.NewServer(options.serverOpts...)
	certPool := x509.NewCertPool()
	x509Cert, err := x509.ParseCertificate(options.cert.Certificate[0])
	if err != nil {
		log.Fatalf("grpcutil: failed parsing x509 certificate: %+v, ", err)
	}
	certPool.AddCert(x509Cert)
	clientCreds := credentials.NewClientTLSFromCert(certPool, "")

	clientOpts := []grpc.DialOption{
		grpc.WithDecompressor(options.decompressor),
		grpc.WithTransportCredentials(clientCreds),
		grpc.WithUserAgent("grpc-gw"),
	}

	address := "localhost:" + options.port
	clientConn, err := grpc.Dial(address, clientOpts...)
	if err != nil {
		log.Fatalf("failed to connect to local gRPC server: %v", err)
	}

	gwMuxer := runtime.NewServeMux()

	serviceBinding := ServiceBinding{
		GRPCServer:        server,
		GRPCGatewayClient: clientConn,
		GRPCGatewayMuxer:  gwMuxer,
	}

	log.Println("registering GRPC services...")
	for _, register := range options.services {
		if err := register(serviceBinding); err != nil {
			log.Fatalf("failed to register service: %+v", err)
		}
	}

	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		contentType := r.Header.Get("Content-Type")
		if r.ProtoMajor == 2 && strings.Contains(contentType, "application/grpc") {
			server.ServeHTTP(w, r)
			return
		}

		for _, p := range options.skipPaths {
			if strings.HasPrefix(r.URL.Path, p) {
				h.ServeHTTP(w, r)
				return
			}
		}

		accept := r.Header.Get("Accept")
		if strings.Contains(contentType, "application/json") ||
			strings.Contains(accept, "application/json") {
			gwMuxer.ServeHTTP(w, r)
			return
		}
		h.ServeHTTP(w, r)
	})
}

{'fraction_non_alphanumeric': 0.11159297651787603, 'fraction_numerical': 0.005711867992384176, 'mean_word_length': 4.091545503500269, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6578528', 'n_tokens_mistral': 3217, 'n_tokens_neox': 3071, 'n_words': 845}