added interceptors, go http pooling & removed unused settings

Author: xiazcy

gremlin-go/driver/client.go

@@ -21,25 +21,15 @@ package gremlingo
 
 import (
 	"crypto/tls"
-	"fmt"
 	"reflect"
 	"runtime"
 	"time"
 
 	"golang.org/x/text/language"
 )
 
-const keepAliveIntervalDefault = 5 * time.Second
-const writeDeadlineDefault = 3 * time.Second
 const connectionTimeoutDefault = 5 * time.Second
 
-// ReadBufferSize and WriteBufferSize specify I/O buffer sizes in bytes. The default is 64KB.
-// If a buffer size is set zero, then the transporter default size is used. The I/O buffer
-// sizes do not limit the size of the messages that can be sent or received.
-const readBufferSizeDefault = 65536  // 64KB
-const maxReadBufferSize = 1073741824 // 1GB - Go's maximum per Read() call
-const writeBufferSizeDefault = 65536 // 64KB
-
 // ClientSettings is used to modify a Client's settings on initialization.
 type ClientSettings struct {
 	TraversalSource   string
@@ -48,16 +38,15 @@ type ClientSettings struct {
 	Language          language.Tag
 	AuthInfo          AuthInfoProvider
 	TlsConfig         *tls.Config
-	KeepAliveInterval time.Duration
-	WriteDeadline     time.Duration
 	ConnectionTimeout time.Duration
 	EnableCompression bool
-	ReadBufferSize    int
-	WriteBufferSize   int
 
 	// Maximum number of concurrent connections. Default: number of runtime processors
 	MaximumConcurrentConnections int
 	EnableUserAgentOnConnect     bool
+
+	// RequestInterceptors are functions that modify HTTP requests before sending.
+	RequestInterceptors []RequestInterceptor
 }
 
 // protocol defines the interface for HTTP communication with Gremlin server
@@ -85,40 +74,33 @@ func NewClient(url string, configurations ...func(settings *ClientSettings)) (*C
 		Language:                 language.English,
 		AuthInfo:                 &AuthInfo{},
 		TlsConfig:                &tls.Config{},
-		KeepAliveInterval:        keepAliveIntervalDefault,
-		WriteDeadline:            writeDeadlineDefault,
 		ConnectionTimeout:        connectionTimeoutDefault,
 		EnableCompression:        false,
 		EnableUserAgentOnConnect: true,
-		ReadBufferSize:           readBufferSizeDefault,
-		WriteBufferSize:          writeBufferSizeDefault,
 
 		MaximumConcurrentConnections: runtime.NumCPU(),
 	}
 	for _, configuration := range configurations {
 		configuration(settings)
 	}
 
-	if settings.ReadBufferSize > maxReadBufferSize {
-		return nil, fmt.Errorf("readBufferSize %d exceeds maximum of %d bytes", settings.ReadBufferSize, maxReadBufferSize)
-	}
-
 	connSettings := &connectionSettings{
 		authInfo:                 settings.AuthInfo,
 		tlsConfig:                settings.TlsConfig,
-		keepAliveInterval:        settings.KeepAliveInterval,
-		writeDeadline:            settings.WriteDeadline,
 		connectionTimeout:        settings.ConnectionTimeout,
 		enableCompression:        settings.EnableCompression,
-		readBufferSize:           settings.ReadBufferSize,
-		writeBufferSize:          settings.WriteBufferSize,
 		enableUserAgentOnConnect: settings.EnableUserAgentOnConnect,
 	}
 
 	logHandler := newLogHandler(settings.Logger, settings.LogVerbosity, settings.Language)
 
 	conn := newHttpConnection(logHandler, url, connSettings)
 
+	// Add user-provided interceptors
+	for _, interceptor := range settings.RequestInterceptors {
+		conn.AddInterceptor(interceptor)
+	}
+
 	client := &Client{
 		url:                url,
 		traversalSource:    settings.TraversalSource,

gremlin-go/driver/connection.go

@@ -27,11 +27,7 @@ import (
 type connectionSettings struct {
 	authInfo                 AuthInfoProvider
 	tlsConfig                *tls.Config
-	keepAliveInterval        time.Duration
-	writeDeadline            time.Duration
 	connectionTimeout        time.Duration
 	enableCompression        bool
-	readBufferSize           int
-	writeBufferSize          int
 	enableUserAgentOnConnect bool
 }

gremlin-go/driver/connection_test.go

@@ -58,13 +58,9 @@ func newDefaultConnectionSettings() *connectionSettings {
 	return &connectionSettings{
 		authInfo:                 &AuthInfo{},
 		tlsConfig:                &tls.Config{},
-		keepAliveInterval:        keepAliveIntervalDefault,
-		writeDeadline:            writeDeadlineDefault,
 		connectionTimeout:        connectionTimeoutDefault,
 		enableCompression:        false,
 		enableUserAgentOnConnect: true,
-		readBufferSize:           readBufferSizeDefault,
-		writeBufferSize:          writeBufferSizeDefault,
 	}
 }
 
@@ -275,14 +271,50 @@ func TestConnection(t *testing.T) {
 	testBasicAuthAuthInfo := getBasicAuthInfo()
 	testBasicAuthTlsConfig := &tls.Config{InsecureSkipVerify: true}
 
+	// this test is used to test the ws->http POC changes via manual execution with a local TP 4.0 gremlin server running on 8182
+	t.Run("Test client.submit()", func(t *testing.T) {
+		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
+
+		tlsConf := tls.Config{
+			InsecureSkipVerify: true,
+		}
+
+		client, err := NewClient(testNoAuthUrl,
+			//client, err := NewClient(noAuthSslUrl,
+			func(settings *ClientSettings) {
+				settings.TlsConfig = &tlsConf
+				settings.AuthInfo = testNoAuthAuthInfo
+				settings.EnableCompression = true
+				settings.TraversalSource = testServerModernGraphAlias
+			})
+		assert.Nil(t, err)
+		assert.NotNil(t, client)
+		defer client.Close()
+
+		// synchronous
+		for i := 0; i < 5; i++ {
+			submitCount(i, client, t)
+		}
+
+		// async
+		var wg sync.WaitGroup
+		for i := 0; i < 5; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				submitCount(i, client, t)
+			}(i)
+		}
+		wg.Wait()
+	})
+
 	t.Run("Test client.submit() with concurrency", func(t *testing.T) {
 		skipTestsIfNotEnabled(t, integrationTestSuiteName, testNoAuthEnable)
 
 		client, err := NewClient(testNoAuthUrl,
 			func(settings *ClientSettings) {
 				settings.TlsConfig = testNoAuthTlsConfig
 				settings.AuthInfo = testNoAuthAuthInfo
-				settings.WriteBufferSize = 1024
 				settings.EnableCompression = true
 				settings.TraversalSource = testServerModernGraphAlias
 			})

gremlin-go/driver/driverRemoteConnection.go

@@ -35,20 +35,15 @@ type DriverRemoteConnectionSettings struct {
 	Language                 language.Tag
 	AuthInfo                 AuthInfoProvider
 	TlsConfig                *tls.Config
-	KeepAliveInterval        time.Duration
-	WriteDeadline            time.Duration
 	ConnectionTimeout        time.Duration
 	EnableCompression        bool
 	EnableUserAgentOnConnect bool
-	ReadBufferSize           int
-	WriteBufferSize          int
 
-	// Minimum amount of concurrent active traversals on a connection to trigger creation of a new connection
-	NewConnectionThreshold int
 	// Maximum number of concurrent connections. Default: number of runtime processors
 	MaximumConcurrentConnections int
-	// Initial amount of instantiated connections. Default: 1
-	InitialConcurrentConnections int
+
+	// RequestInterceptors are functions that modify HTTP requests before sending.
+	RequestInterceptors []RequestInterceptor
 }
 
 // DriverRemoteConnection is a remote connection.
@@ -72,13 +67,9 @@ func NewDriverRemoteConnection(
 		Language:                 language.English,
 		AuthInfo:                 &AuthInfo{},
 		TlsConfig:                &tls.Config{},
-		KeepAliveInterval:        keepAliveIntervalDefault,
-		WriteDeadline:            writeDeadlineDefault,
 		ConnectionTimeout:        connectionTimeoutDefault,
 		EnableCompression:        false,
 		EnableUserAgentOnConnect: true,
-		ReadBufferSize:           readBufferSizeDefault,
-		WriteBufferSize:          writeBufferSizeDefault,
 
 		MaximumConcurrentConnections: runtime.NumCPU(),
 	}
@@ -89,19 +80,20 @@ func NewDriverRemoteConnection(
 	connSettings := &connectionSettings{
 		authInfo:                 settings.AuthInfo,
 		tlsConfig:                settings.TlsConfig,
-		keepAliveInterval:        settings.KeepAliveInterval,
-		writeDeadline:            settings.WriteDeadline,
 		connectionTimeout:        settings.ConnectionTimeout,
 		enableCompression:        settings.EnableCompression,
-		readBufferSize:           settings.ReadBufferSize,
-		writeBufferSize:          settings.WriteBufferSize,
 		enableUserAgentOnConnect: settings.EnableUserAgentOnConnect,
 	}
 
 	logHandler := newLogHandler(settings.Logger, settings.LogVerbosity, settings.Language)
 
 	conn := newHttpConnection(logHandler, url, connSettings)
 
+	// Add user-provided interceptors
+	for _, interceptor := range settings.RequestInterceptors {
+		conn.AddInterceptor(interceptor)
+	}
+
 	client := &Client{
 		url:                url,
 		traversalSource:    settings.TraversalSource,