combined httpProtocol & Transport

Author: xiazcy

gremlin-go/driver/client.go

@@ -60,13 +60,19 @@ type ClientSettings struct {
 	EnableUserAgentOnConnect     bool
 }
 
+// protocol defines the interface for HTTP communication with Gremlin server
+type protocol interface {
+	send(request *request) (ResultSet, error)
+	close()
+}
+
 // Client is used to connect and interact with a Gremlin-supported server.
 type Client struct {
 	url                string
 	traversalSource    string
 	logHandler         *logHandler
 	connectionSettings *connectionSettings
-	httpProtocol       *httpProtocol
+	protocol           protocol
 }
 
 // NewClient creates a Client and configures it with the given parameters.
@@ -111,14 +117,14 @@ func NewClient(url string, configurations ...func(settings *ClientSettings)) (*C
 
 	logHandler := newLogHandler(settings.Logger, settings.LogVerbosity, settings.Language)
 
-	httpProt := newHttpProtocol(logHandler, url, connSettings)
+	conn := newHttpConnection(logHandler, url, connSettings)
 
 	client := &Client{
 		url:                url,
 		traversalSource:    settings.TraversalSource,
 		logHandler:         logHandler,
 		connectionSettings: connSettings,
-		httpProtocol:       httpProt,
+		protocol:           conn,
 	}
 
 	return client, nil
@@ -142,7 +148,7 @@ func (client *Client) SubmitWithOptions(traversalString string, requestOptions R
 
 	// TODO interceptors (ie. auth)
 
-	rs, err := client.httpProtocol.send(&request)
+	rs, err := client.protocol.send(&request)
 	return rs, err
 }
 
@@ -171,7 +177,7 @@ func (client *Client) submitGremlinLang(gremlinLang *GremlinLang) (ResultSet, er
 	}
 
 	request := MakeStringRequest(gremlinLang.GetGremlin(), client.traversalSource, requestOptionsBuilder.Create())
-	return client.httpProtocol.send(&request)
+	return client.protocol.send(&request)
 }
 
 func applyOptionsConfig(builder *RequestOptionsBuilder, config map[string]interface{}) *RequestOptionsBuilder {

gremlin-go/driver/driverRemoteConnection.go

@@ -100,14 +100,14 @@ func NewDriverRemoteConnection(
 
 	logHandler := newLogHandler(settings.Logger, settings.LogVerbosity, settings.Language)
 
-	httpProt := newHttpProtocol(logHandler, url, connSettings)
+	conn := newHttpConnection(logHandler, url, connSettings)
 
 	client := &Client{
 		url:                url,
 		traversalSource:    settings.TraversalSource,
 		logHandler:         logHandler,
 		connectionSettings: connSettings,
-		httpProtocol:       httpProt,
+		protocol:           conn,
 	}
 
 	return &DriverRemoteConnection{client: client, isClosed: false, settings: settings}, nil

gremlin-go/driver/httpConnection.go

@@ -0,0 +1,196 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package gremlingo
+
+import (
+	"bytes"
+	"compress/zlib"
+	"io"
+	"net/http"
+	"time"
+)
+
+// httpConnection handles HTTP request/response for Gremlin queries
+type httpConnection struct {
+	url          string
+	httpClient   *http.Client
+	connSettings *connectionSettings
+	logHandler   *logHandler
+	serializer   *GraphBinarySerializer
+}
+
+// Connection pool defaults aligned with Java driver
+const (
+	defaultMaxConnsPerHost     = 128               // Java: ConnectionPool.MAX_POOL_SIZE
+	defaultMaxIdleConnsPerHost = 8                 // Keep some connections warm
+	defaultIdleConnTimeout     = 180 * time.Second // Java: CONNECTION_IDLE_TIMEOUT_MILLIS
+	defaultConnectionTimeout   = 15 * time.Second  // Java: CONNECTION_SETUP_TIMEOUT_MILLIS
+)
+
+func newHttpConnection(handler *logHandler, url string, connSettings *connectionSettings) *httpConnection {
+	timeout := connSettings.connectionTimeout
+	if timeout == 0 {
+		timeout = defaultConnectionTimeout
+	}
+
+	transport := &http.Transport{
+		TLSClientConfig:     connSettings.tlsConfig,
+		MaxConnsPerHost:     defaultMaxConnsPerHost,
+		MaxIdleConnsPerHost: defaultMaxIdleConnsPerHost,
+		IdleConnTimeout:     defaultIdleConnTimeout,
+		DisableCompression:  !connSettings.enableCompression,
+	}
+
+	return &httpConnection{
+		url:          url,
+		httpClient:   &http.Client{Transport: transport, Timeout: timeout},
+		connSettings: connSettings,
+		logHandler:   handler,
+		serializer:   newGraphBinarySerializer(handler),
+	}
+}
+
+// send sends request and streams results directly to ResultSet
+func (c *httpConnection) send(req *request) (ResultSet, error) {
+	rs := newChannelResultSet()
+
+	data, err := c.serializer.SerializeMessage(req)
+	if err != nil {
+		rs.Close()
+		return rs, err
+	}
+
+	go c.executeAndStream(data, rs)
+
+	return rs, nil
+}
+
+func (c *httpConnection) executeAndStream(data []byte, rs ResultSet) {
+	defer rs.Close()
+
+	req, err := http.NewRequest(http.MethodPost, c.url, bytes.NewReader(data))
+	if err != nil {
+		c.logHandler.logf(Error, failedToSendRequest, err.Error())
+		rs.setError(err)
+		return
+	}
+
+	c.setHeaders(req)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		c.logHandler.logf(Error, failedToSendRequest, err.Error())
+		rs.setError(err)
+		return
+	}
+	defer resp.Body.Close()
+
+	reader, zlibReader, err := c.getReader(resp)
+	if err != nil {
+		c.logHandler.logf(Error, failedToReceiveResponse, err.Error())
+		rs.setError(err)
+		return
+	}
+	if zlibReader != nil {
+		defer zlibReader.Close()
+	}
+
+	c.streamToResultSet(reader, rs)
+}
+
+func (c *httpConnection) setHeaders(req *http.Request) {
+	req.Header.Set("Content-Type", graphBinaryMimeType)
+	req.Header.Set("Accept", graphBinaryMimeType)
+
+	if c.connSettings.enableUserAgentOnConnect {
+		req.Header.Set(userAgentHeader, userAgent)
+	}
+	if c.connSettings.enableCompression {
+		req.Header.Set("Accept-Encoding", "deflate")
+	}
+	if c.connSettings.authInfo != nil {
+		if headers := c.connSettings.authInfo.GetHeader(); headers != nil {
+			for k, vals := range headers {
+				for _, v := range vals {
+					req.Header.Add(k, v)
+				}
+			}
+		}
+		if ok, user, pass := c.connSettings.authInfo.GetBasicAuth(); ok {
+			req.SetBasicAuth(user, pass)
+		}
+	}
+}
+
+func (c *httpConnection) getReader(resp *http.Response) (io.Reader, io.Closer, error) {
+	if resp.Header.Get("Content-Encoding") == "deflate" {
+		zr, err := zlib.NewReader(resp.Body)
+		if err != nil {
+			return nil, nil, err
+		}
+		return zr, zr, nil
+	}
+	return resp.Body, nil, nil
+}
+
+func (c *httpConnection) streamToResultSet(reader io.Reader, rs ResultSet) {
+	d := newStreamingDeserializer(reader)
+	if err := d.readHeader(); err != nil {
+		if err != io.EOF {
+			c.logHandler.logf(Error, failedToReceiveResponse, err.Error())
+			rs.setError(err)
+		}
+		return
+	}
+
+	for {
+		obj, err := d.readFullyQualified()
+		if err != nil {
+			if err != io.EOF {
+				c.logHandler.logf(Error, failedToReceiveResponse, err.Error())
+				rs.setError(err)
+			}
+			return
+		}
+
+		if marker, ok := obj.(Marker); ok && marker == EndOfStream() {
+			code, msg, exc, err := d.readStatus()
+			if err != nil {
+				c.logHandler.logf(Error, failedToReceiveResponse, err.Error())
+				rs.setError(err)
+				return
+			}
+			if code != 200 && code != 0 {
+				if exc != "" {
+					rs.setError(newError(err0502ResponseHandlerReadLoopError, exc, code))
+				} else {
+					rs.setError(newError(err0502ResponseHandlerReadLoopError, msg, code))
+				}
+			}
+			return
+		}
+
+		rs.Channel() <- &Result{obj}
+	}
+}
+
+func (c *httpConnection) close() {
+	c.httpClient.CloseIdleConnections()
+}