Add ngx binary response handling for GetItems (#21)

go.mod

@@ -15,4 +15,5 @@ require (
 	github.com/valyala/fasthttp v1.2.0
 	go.uber.org/atomic v1.3.2 // indirect
 	go.uber.org/multierr v1.1.0 // indirect
+	zombiezen.com/go/capnproto2 v2.17.0+incompatible
 )

go.sum

@@ -1,3 +1,5 @@
+github.com/capnproto/go-capnproto2 v2.17.0+incompatible h1:vPbYlc2CBNdjzOMzHfwo7TbFNRBDaRKitlWiRs1riTw=
+github.com/capnproto/go-capnproto2 v2.17.0+incompatible/go.mod h1:T3/pxeK0qevFRlAASYZe90Ozs+JmlQTNY+VLc6+lJHw=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -36,6 +38,9 @@ go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
+golang.org/x/net v0.0.0-20180911220305-26e67e76b6c3 h1:czFLhve3vsQetD6JOJ8NZZvGQIXlnN3/yXxbT6/awxI=
 golang.org/x/net v0.0.0-20180911220305-26e67e76b6c3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223 h1:DH4skfRX4EBpamg7iV4ZlCpblAHI6s6TDM39bFZumv8=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+zombiezen.com/go/capnproto2 v2.17.0+incompatible h1:sIoKPFGNlM38Qh+PBLa9Wzg1j99oInS/Qlk+5N/CHa4=
+zombiezen.com/go/capnproto2 v2.17.0+incompatible/go.mod h1:XO5Pr2SbXgqZwn0m0Ru54QBqpOf4K5AYBO+8LAOBQEQ=

pkg/dataplane/http/context.go

@@ -16,13 +16,15 @@ import (
 	"strings"
 	"sync/atomic"
 	"time"
+	"io"
 
 	"github.com/v3io/v3io-go/pkg/dataplane"
 	"github.com/v3io/v3io-go/pkg/errors"
-
+	"github.com/v3io/v3io-go/pkg/dataplane/schemas/node/common"
 	"github.com/nuclio/errors"
 	"github.com/nuclio/logger"
 	"github.com/valyala/fasthttp"
+	"zombiezen.com/go/capnproto2"
 )
 
 // TODO: Request should have a global pool
@@ -206,6 +208,11 @@ func (c *context) GetItem(getItemInput *v3io.GetItemInput,
 	return c.sendRequestToWorker(getItemInput, context, responseChan)
 }
 
+type attributeValuesSection struct {
+	accumulatedPreviousSectionsLength int
+	data node_common_capnp.VnObjectAttributeValuePtr_List
+}
+
 // GetItemSync
 func (c *context) GetItemSync(getItemInput *v3io.GetItemInput) (*v3io.Response, error) {
 
@@ -256,6 +263,7 @@ func (c *context) GetItems(getItemsInput *v3io.GetItemsInput,
 	return c.sendRequestToWorker(getItemsInput, context, responseChan)
 }
 
+
 // GetItemSync
 func (c *context) GetItemsSync(getItemsInput *v3io.GetItemsInput) (*v3io.Response, error) {
 
@@ -308,55 +316,23 @@ func (c *context) GetItemsSync(getItemsInput *v3io.GetItemsInput) (*v3io.Respons
 		"PUT",
 		getItemsInput.Path,
 		"",
-		getItemsHeaders,
+		getItemsHeadersCapnp,
 		marshalledBody,
 		false)
 
 	if err != nil {
 		return nil, err
 	}
 
-	c.logger.DebugWithCtx(getItemsInput.Ctx, "Body", "body", string(response.Body()))
-
-	getItemsResponse := struct {
-		Items            []map[string]map[string]interface{}
-		NextMarker       string
-		LastItemIncluded string
-	}{}
-
-	// unmarshal the body into an ad hoc structure
-	err = json.Unmarshal(response.Body(), &getItemsResponse)
-	if err != nil {
-		return nil, err
-	}
-
-	//validate getItems response to avoid infinite loop
-	if getItemsResponse.LastItemIncluded != "TRUE" && (getItemsResponse.NextMarker == "" || getItemsResponse.NextMarker == getItemsInput.Marker) {
-		errMsg := fmt.Sprintf("Invalid getItems response: lastItemIncluded=false and nextMarker='%s', "+
-			"startMarker='%s', probably due to object size bigger than 2M. Query is: %+v", getItemsResponse.NextMarker, getItemsInput.Marker, getItemsInput)
-		c.logger.Warn(errMsg)
-	}
-
-	getItemsOutput := v3io.GetItemsOutput{
-		NextMarker: getItemsResponse.NextMarker,
-		Last:       getItemsResponse.LastItemIncluded == "TRUE",
-	}
-
-	// iterate through the items and decode them
-	for _, typedItem := range getItemsResponse.Items {
-
-		item, err := c.decodeTypedAttributes(typedItem)
-		if err != nil {
-			return nil, err
-		}
+	contentType := string(response.HeaderPeek("Content-Type"))
 
-		getItemsOutput.Items = append(getItemsOutput.Items, item)
+	if contentType != "application/octet-capnp" {
+		c.logger.DebugWithCtx(getItemsInput.Ctx, "Body", "body", string(response.Body()))
+		response.Output, err = c.getItemsParseJSONResponse(response, getItemsInput)
+	} else {
+		response.Output, err = c.getItemsParseCAPNPResponse(response)
 	}
-
-	// attach the output to the response
-	response.Output = &getItemsOutput
-
-	return response, nil
+	return response, err
 }
 
 // PutItem
@@ -1141,3 +1117,205 @@ func (c *context) workerEntry(workerIndex int) {
 		request.ResponseChan <- &request.RequestResponse.Response
 	}
 }
+
+func readAllCapnpMessages(reader io.Reader) ([]*capnp.Message){
+	var capnpMessages []*capnp.Message
+	for {
+		msg, err := capnp.NewDecoder(reader).Decode()
+		if err != nil {
+			break
+		}
+		capnpMessages = append(capnpMessages, msg)
+	}
+	return capnpMessages
+}
+
+func getSectionAndIndex(values []attributeValuesSection, idx int) (section int, resIdx int){
+	if len(values) == 1 {
+		return 0, idx
+	}
+	for i := 1; i<len(values); i++ {
+		if values[i].accumulatedPreviousSectionsLength > idx {
+			return i,idx - values[i-1].accumulatedPreviousSectionsLength
+		}
+	}
+	return 0, idx
+}
+
+func decodeCapnpAttributes(keyValues node_common_capnp.VnObjectItemsGetMappedKeyValuePair_List, values []attributeValuesSection, attributeNames []string) (map[string]interface{}, error) {
+	attributes := map[string]interface{}{}
+	for j := 0; j < keyValues.Len(); j++ {
+		attrPtr := keyValues.At(j)
+		valIdx := int(attrPtr.ValueMapIndex())
+		attrIdx := int(attrPtr.KeyMapIndex())
+
+		attributeName := attributeNames[attrIdx]
+		sectIdx, valIdx := getSectionAndIndex(values, valIdx)
+		value, err := values[sectIdx].data.At(valIdx).Value()
+		if err != nil {
+			return attributes, errors.Wrapf(err,"values[%d].data.At(%d).Value",sectIdx, valIdx )
+		}
+		switch value.Which() {
+		case node_common_capnp.ExtAttrValue_Which_qword:
+			attributes[attributeName] = int(value.Qword())
+		case node_common_capnp.ExtAttrValue_Which_uqword:
+			attributes[attributeName] = int(value.Uqword())
+		case node_common_capnp.ExtAttrValue_Which_blob:
+			attributes[attributeName], err = value.Blob()
+		case node_common_capnp.ExtAttrValue_Which_str:
+			attributes[attributeName], err = value.Str()
+		case node_common_capnp.ExtAttrValue_Which_dfloat:
+			attributes[attributeName] = value.Dfloat()
+		case node_common_capnp.ExtAttrValue_Which_boolean:
+			attributes[attributeName] = value.Boolean()
+		case node_common_capnp.ExtAttrValue_Which_notExists:
+			{}
+		default:
+			return attributes, errors.Errorf("getItemsCapnp: %s type for %s attribute is not expected", value.Which().String(), attributeName)
+		}
+	}
+	return attributes, nil
+}
+
+func (c *context) getItemsParseJSONResponse(response *v3io.Response, getItemsInput *v3io.GetItemsInput) (*v3io.GetItemsOutput, error){
+
+	getItemsResponse := struct {
+		Items            []map[string]map[string]interface{}
+		NextMarker       string
+		LastItemIncluded string
+	}{}
+
+	// unmarshal the body into an ad hoc structure
+	err := json.Unmarshal(response.Body(), &getItemsResponse)
+	if err != nil {
+		return nil, err
+	}
+
+	//validate getItems response to avoid infinite loop
+	if getItemsResponse.LastItemIncluded != "TRUE" && (getItemsResponse.NextMarker == "" || getItemsResponse.NextMarker == getItemsInput.Marker) {
+		errMsg := fmt.Sprintf("Invalid getItems response: lastItemIncluded=false and nextMarker='%s', "+
+			"startMarker='%s', probably due to object size bigger than 2M. Query is: %+v", getItemsResponse.NextMarker, getItemsInput.Marker, getItemsInput)
+		c.logger.Warn(errMsg)
+	}
+
+	getItemsOutput := v3io.GetItemsOutput{
+		NextMarker: getItemsResponse.NextMarker,
+		Last:       getItemsResponse.LastItemIncluded == "TRUE",
+	}
+
+	// iterate through the items and decode them
+	for _, typedItem := range getItemsResponse.Items {
+
+		item, err := c.decodeTypedAttributes(typedItem)
+		if err != nil {
+			return nil, err
+		}
+
+		getItemsOutput.Items = append(getItemsOutput.Items, item)
+	}
+	// attach the output to the response
+	return &getItemsOutput, nil
+}
+
+func (c *context) getItemsParseCAPNPResponse(response *v3io.Response) (*v3io.GetItemsOutput, error){
+	responseBodyReader  := bytes.NewReader(response.Body())
+	capnpSections := readAllCapnpMessages(responseBodyReader )
+	if len(capnpSections) < 2 {
+		return  nil, errors.Errorf("getItemsCapnp: Got only %v capnp sections. Expecting at least 2", len(capnpSections))
+	}
+	cookie := string(response.HeaderPeek("X-v3io-cookie"))
+	getItemsOutput := v3io.GetItemsOutput{
+		NextMarker: cookie,
+		Last:       len(cookie) == 0,
+	}
+	if len(capnpSections) < 2 {
+		return nil, errors.Errorf("getItemsCapnp: Got only %v capnp sections. Expecting at least 2", len(capnpSections))
+	}
+
+	metadataPayload, err := node_common_capnp.ReadRootVnObjectItemsGetResponseMetadataPayload(capnpSections[len(capnpSections) - 1])
+	if err != nil {
+		return nil, errors.Wrap(err,"ReadRootVnObjectItemsGetResponseMetadataPayload")
+	}
+	//Keys
+	attributeMap, err := metadataPayload.KeyMap()
+	 if err != nil {
+		return nil, errors.Wrap(err,"metadataPayload.KeyMap")
+	}
+	attributeMapNames,err := attributeMap.Names()
+	 if err != nil {
+		return nil, errors.Wrap(err,"attributeMap.Names")
+	}
+	attributeNamesPtr,err  := attributeMapNames.Arr()
+	if err != nil {
+		return nil, errors.Wrap(err,"attributeMapNames.Arr")
+	}
+	//Values
+	valueMap, err := metadataPayload.ValueMap()
+	if err != nil {
+		return nil, errors.Wrap(err,"metadataPayload.ValueMap")
+	}
+	values, err := valueMap.Values()
+	if err != nil {
+		return nil, errors.Wrap(err,"valueMap.Values")
+	}
+
+	// Items
+	items, err := metadataPayload.Items()
+	if err != nil {
+		return nil, errors.Wrap(err,"metadataPayload.Items")
+	}
+	valuesSections :=  make([]attributeValuesSection , len(capnpSections) - 1)
+
+	accLength := 0
+	//Additional data sections "in between"
+	for capnpSectionIndex:= 1;capnpSectionIndex < len(capnpSections) - 1; capnpSectionIndex++ {
+		data, err := node_common_capnp.ReadRootVnObjectAttributeValueMap(capnpSections[capnpSectionIndex])
+		if err != nil {
+			return nil, errors.Wrap(err,"node_common_capnp.ReadRootVnObjectAttributeValueMap")
+		}
+		dv, err := data.Values()
+		if err != nil {
+			return nil, errors.Wrap(err,"data.Values")
+		}
+		accLength = accLength + dv.Len()  
+		valuesSections[capnpSectionIndex-1].data = dv
+		valuesSections[capnpSectionIndex-1].accumulatedPreviousSectionsLength = accLength
+	}
+	accLength = accLength + values.Len()  
+	valuesSections[len(capnpSections) - 2].data = values
+	valuesSections[len(capnpSections) - 2].accumulatedPreviousSectionsLength = accLength
+
+	//Read in all the attribute names
+	attributeNamesNumber := attributeNamesPtr.Len()
+	attributeNames := make([]string,attributeNamesNumber)
+	for attributeIndex := 0; attributeIndex < attributeNamesNumber; attributeIndex++ {
+		attributeNames[attributeIndex], err = attributeNamesPtr.At(attributeIndex).Str()
+		if err != nil {
+			return nil, errors.Wrapf(err,"attributeNamesPtr.At(%d) size %d", attributeIndex, attributeNamesNumber)
+		}
+	}
+
+	// iterate through the items and decode them
+	for itemIndex := 0; itemIndex < items.Len(); itemIndex++ {
+		itemPtr := items.At(itemIndex)
+		item, err := itemPtr.Item()
+		if err != nil {
+			return nil, errors.Wrap(err,"itemPtr.Item")
+		}
+		name, err := item.Name()
+		if err != nil {
+			return nil, errors.Wrap(err,"item.Name")
+		}
+		itemAttributes, err := item.Attrs()
+		if err != nil {
+			return nil, errors.Wrap(err,"item.Attrs")
+		}
+		ditem, err := decodeCapnpAttributes(itemAttributes, valuesSections, attributeNames)
+		if err != nil {
+			return nil, errors.Wrap(err,"decodeCapnpAttributes")
+		}
+		ditem["__name"] = name
+		getItemsOutput.Items = append(getItemsOutput.Items, ditem)
+	}
+	return &getItemsOutput, nil
+}

pkg/dataplane/http/headers.go

@@ -36,6 +36,13 @@ var getItemsHeaders = map[string]string{
 	"X-v3io-function": getItemsFunctionName,
 }
 
+// headers for update item
+var getItemsHeadersCapnp = map[string]string{
+	"Content-Type":    "application/json",
+	"X-v3io-response-content-type": "capnp",
+	"X-v3io-function": getItemsFunctionName,
+}
+
 // headers for create stream
 var createStreamHeaders = map[string]string{
 	"Content-Type":    "application/json",

pkg/dataplane/requestresponse.go

@@ -68,6 +68,11 @@ func (r *Response) Body() []byte {
 	return r.HTTPResponse.Body()
 }
 
+func (r *Response) HeaderPeek(key string) []byte {
+	return r.HTTPResponse.Header.Peek(key)
+}
+
+
 func (r *Response) Request() *Request {
 	return &r.RequestResponse.Request
 }

pkg/dataplane/schemas/build

@@ -0,0 +1,12 @@
+#!/bin/bash
+# put the capnpc-go binary in /bin
+export PATH=$PATH:${GOPATH}/bin
+
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/ExtAttrValue.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/StringWrapper.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/TimeSpec.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/VnObjectAttributeKeyMap.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/VnObjectAttributeValueMap.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/VnObjectItemsGetResponse.capnp
+capnp compile -I$GOPATH/src/github.com/iguazio/go-capnproto2/std -I$(pwd) -ogo node/common/VnObjectItemsScanCookie.capnp
+

pkg/dataplane/schemas/clean

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+find . -name *.go -delete

pkg/dataplane/schemas/go.capnp

@@ -0,0 +1,27 @@
+@0xd12a1c51fedd6c88;
+
+annotation package(file) :Text;
+# The Go package name for the generated file.
+
+annotation import(file) :Text;
+# The Go import path that the generated file is accessible from.
+# Used to generate import statements and check if two types are in the
+# same package.
+
+annotation doc(struct, field, enum) :Text;
+# Adds a doc comment to the generated code.
+
+# annotation tag(enumerant) :Text;
+# Changes the string representation of the enum in the generated code.
+
+# annotation notag(enumerant) :Void;
+# Removes the string representation of the enum in the generated code.
+
+# annotation customtype(field) :Text;
+# OBSOLETE, not used by code generator.
+
+# annotation name(struct, field, union, enum, enumerant, interface, method, param, annotation, const, group) :Text;
+# Used to rename the element in the generated code.
+
+$package("capnp");
+