From db2a53223d591a396c5d50964d3e675069616af9 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 5 Jul 2018 16:02:18 -0500 Subject: [PATCH] remove internal references from api and http/* --- api.go | 14 ++--- encoding/proto/proto.go | 134 +++++++++++++++++++++++++++++++++++++++- handler.go | 37 +++++++++++ http/client.go | 34 +++++----- http/handler.go | 12 ++-- 5 files changed, 197 insertions(+), 34 deletions(-) diff --git a/api.go b/api.go index 1abe4502c..181d1b3fb 100644 --- a/api.go +++ b/api.go @@ -26,8 +26,6 @@ import ( "strings" "time" - "github.com/gogo/protobuf/proto" - "github.com/pilosa/pilosa/internal" "github.com/pilosa/pilosa/pql" "github.com/pkg/errors" ) @@ -437,8 +435,8 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, if err != nil { return nil, NewBadRequestError(errors.Wrap(err, "read body error")) } - var req internal.BlockDataRequest - if err := proto.Unmarshal(reqBytes, &req); err != nil { + var req BlockDataRequest + if err := api.Serializer.Unmarshal(reqBytes, &req); err != nil { return nil, NewBadRequestError(errors.Wrap(err, "unmarshal body error")) } @@ -448,11 +446,11 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, return nil, ErrFragmentNotFound } - var resp = internal.BlockDataResponse{} + var resp = BlockDataResponse{} resp.RowIDs, resp.ColumnIDs = f.blockData(int(req.Block)) // Encode response. - buf, err := proto.Marshal(&resp) + buf, err := api.Serializer.Marshal(&resp) if err != nil { return nil, errors.Wrap(err, "merge block response encoding error") } @@ -657,7 +655,7 @@ func (api *API) FieldAttrDiff(ctx context.Context, indexName string, fieldName s } // Import bulk imports data into a particular index,field,shard. -func (api *API) Import(ctx context.Context, req internal.ImportRequest) error { +func (api *API) Import(ctx context.Context, req *ImportRequest) error { if err := api.validate(apiImport); err != nil { return errors.Wrap(err, "validating api method") } @@ -686,7 +684,7 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error { } // ImportValue bulk imports values into a particular field. -func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest) error { +func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest) error { if err := api.validate(apiImportValue); err != nil { return errors.Wrap(err, "validating api method") } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 003e9cd94..ed751247e 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -178,7 +178,46 @@ func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error { } decodeQueryResponse(msg, mt) return nil - + case *pilosa.ImportRequest: + msg := &internal.ImportRequest{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ImportRequest") + } + decodeImportRequest(msg, mt) + return nil + case *pilosa.ImportValueRequest: + msg := &internal.ImportValueRequest{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ImportValueRequest") + } + decodeImportValueRequest(msg, mt) + return nil + case *pilosa.ImportResponse: + msg := &internal.ImportResponse{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ImportResponse") + } + decodeImportResponse(msg, mt) + return nil + case *pilosa.BlockDataRequest: + msg := &internal.BlockDataRequest{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling BlockDataRequest") + } + decodeBlockDataRequest(msg, mt) + return nil + case *pilosa.BlockDataResponse: + msg := &internal.BlockDataResponse{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling BlockDataResponse") + } + decodeBlockDataResponse(msg, mt) + return nil default: panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m)) } @@ -224,10 +263,66 @@ func encodeToProto(m pilosa.Message) proto.Message { return encodeQueryRequest(mt) case *pilosa.QueryResponse: return encodeQueryResponse(mt) + case *pilosa.ImportRequest: + return encodeImportRequest(mt) + case *pilosa.ImportValueRequest: + return encodeImportValueRequest(mt) + case *pilosa.ImportResponse: + return encodeImportResponse(mt) + case *pilosa.BlockDataRequest: + return encodeBlockDataRequest(mt) + case *pilosa.BlockDataResponse: + return encodeBlockDataResponse(mt) } return nil } +func encodeBlockDataRequest(m *pilosa.BlockDataRequest) *internal.BlockDataRequest { + return &internal.BlockDataRequest{ + Index: m.Index, + Field: m.Field, + View: m.View, + Shard: m.Shard, + Block: m.Block, + } +} +func encodeBlockDataResponse(m *pilosa.BlockDataResponse) *internal.BlockDataResponse { + return &internal.BlockDataResponse{ + RowIDs: m.RowIDs, + ColumnIDs: m.ColumnIDs, + } +} + +func encodeImportResponse(m *pilosa.ImportResponse) *internal.ImportResponse { + return &internal.ImportResponse{ + Err: m.Err, + } +} + +func encodeImportRequest(m *pilosa.ImportRequest) *internal.ImportRequest { + return &internal.ImportRequest{ + Index: m.Index, + Field: m.Field, + Shard: m.Shard, + RowIDs: m.RowIDs, + ColumnIDs: m.ColumnIDs, + RowKeys: m.RowKeys, + ColumnKeys: m.ColumnKeys, + Timestamps: m.Timestamps, + } +} + +func encodeImportValueRequest(m *pilosa.ImportValueRequest) *internal.ImportValueRequest { + return &internal.ImportValueRequest{ + Index: m.Index, + Field: m.Field, + Shard: m.Shard, + ColumnIDs: m.ColumnIDs, + ColumnKeys: m.ColumnKeys, + Values: m.Values, + } +} + func encodeQueryRequest(m *pilosa.QueryRequest) *internal.QueryRequest { return &internal.QueryRequest{ Query: m.Query, @@ -690,6 +785,43 @@ func decodeQueryRequest(pb *internal.QueryRequest, m *pilosa.QueryRequest) { m.ExcludeColumns = pb.ExcludeColumns } +func decodeImportRequest(pb *internal.ImportRequest, m *pilosa.ImportRequest) { + m.Index = pb.Index + m.Field = pb.Field + m.Shard = pb.Shard + m.RowIDs = pb.RowIDs + m.ColumnIDs = pb.ColumnIDs + m.RowKeys = pb.RowKeys + m.ColumnKeys = pb.ColumnKeys + m.Timestamps = pb.Timestamps +} + +func decodeImportValueRequest(pb *internal.ImportValueRequest, m *pilosa.ImportValueRequest) { + m.Index = pb.Index + m.Field = pb.Field + m.Shard = pb.Shard + m.ColumnIDs = pb.ColumnIDs + m.ColumnKeys = pb.ColumnKeys + m.Values = pb.Values +} + +func decodeImportResponse(pb *internal.ImportResponse, m *pilosa.ImportResponse) { + m.Err = pb.Err +} + +func decodeBlockDataRequest(pb *internal.BlockDataRequest, m *pilosa.BlockDataRequest) { + m.Index = pb.Index + m.Field = pb.Field + m.View = pb.View + m.Shard = pb.Shard + m.Block = pb.Block +} + +func decodeBlockDataResponse(pb *internal.BlockDataResponse, m *pilosa.BlockDataResponse) { + m.RowIDs = pb.RowIDs + m.ColumnIDs = pb.ColumnIDs +} + func decodeQueryResponse(pb *internal.QueryResponse, m *pilosa.QueryResponse) { m.ColumnAttrSets = make([]*pilosa.ColumnAttrSet, len(pb.ColumnAttrSets)) decodeColumnAttrSets(pb.ColumnAttrSets, m.ColumnAttrSets) diff --git a/handler.go b/handler.go index 1f3d04300..9fc3af368 100644 --- a/handler.go +++ b/handler.go @@ -75,3 +75,40 @@ func (n nopHandler) Close() error { } var NopHandler Handler = nopHandler{} + +type ImportValueRequest struct { + Index string + Field string + Shard uint64 + ColumnIDs []uint64 + ColumnKeys []string + Values []int64 +} + +type ImportRequest struct { + Index string + Field string + Shard uint64 + RowIDs []uint64 + ColumnIDs []uint64 + RowKeys []string + ColumnKeys []string + Timestamps []int64 +} + +type ImportResponse struct { + Err string +} + +type BlockDataRequest struct { + Index string + Field string + View string + Shard uint64 + Block uint64 +} + +type BlockDataResponse struct { + RowIDs []uint64 + ColumnIDs []uint64 +} diff --git a/http/client.go b/http/client.go index d8fb7f7ef..19bf82815 100644 --- a/http/client.go +++ b/http/client.go @@ -29,10 +29,8 @@ import ( "crypto/tls" - "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa" - pilosaproto "github.com/pilosa/pilosa/encoding/proto" - "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/encoding/proto" "github.com/pkg/errors" ) @@ -68,7 +66,7 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, func NewInternalClientFromURI(defaultURI *pilosa.URI, remoteClient *http.Client) *InternalClient { return &InternalClient{ defaultURI: defaultURI, - serializer: pilosaproto.Serializer{}, + serializer: proto.Serializer{}, HTTPClient: remoteClient, } } @@ -282,7 +280,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard return pilosa.ErrFieldRequired } - buf, err := marshalImportPayload(index, field, shard, bits) + buf, err := c.marshalImportPayload(index, field, shard, bits) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -311,7 +309,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, colum return pilosa.ErrFieldRequired } - buf, err := marshalImportPayloadK(index, field, columns) + buf, err := c.marshalImportPayloadK(index, field, columns) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -345,14 +343,14 @@ func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fiel } // marshalImportPayload marshalls the import parameters into a protobuf byte slice. -func marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) { +func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) { // Separate row and column IDs to reduce allocations. rowIDs := Bits(bits).RowIDs() columnIDs := Bits(bits).ColumnIDs() timestamps := Bits(bits).Timestamps() // Marshal data to protobuf. - buf, err := proto.Marshal(&internal.ImportRequest{ + buf, err := c.serializer.Marshal(&pilosa.ImportRequest{ Index: index, Field: field, Shard: shard, @@ -367,14 +365,14 @@ func marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) } // marshalImportPayloadK marshalls the import parameters into a protobuf byte slice. -func marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) { +func (c *InternalClient) marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) { // Separate row and column IDs to reduce allocations. rowKeys := Bits(bits).RowKeys() columnKeys := Bits(bits).ColumnKeys() timestamps := Bits(bits).Timestamps() // Marshal data to protobuf. - buf, err := proto.Marshal(&internal.ImportRequest{ + buf, err := c.serializer.Marshal(&pilosa.ImportRequest{ Index: index, Field: field, RowKeys: rowKeys, @@ -416,8 +414,8 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde return errors.New(string(body)) } - var isresp internal.ImportResponse - if err := proto.Unmarshal(body, &isresp); err != nil { + var isresp pilosa.ImportResponse + if err := c.serializer.Unmarshal(body, &isresp); err != nil { return fmt.Errorf("unmarshal import response: %s", err) } else if s := isresp.Err; s != "" { return errors.New(s) @@ -434,7 +432,7 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s return pilosa.ErrFieldRequired } - buf, err := marshalImportValuePayload(index, field, shard, vals) + buf, err := c.marshalImportValuePayload(index, field, shard, vals) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -456,13 +454,13 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s } // marshalImportValuePayload marshalls the import parameters into a protobuf byte slice. -func marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) { +func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) { // Separate row and column IDs to reduce allocations. columnIDs := FieldValues(vals).ColumnIDs() values := FieldValues(vals).Values() // Marshal data to protobuf. - buf, err := proto.Marshal(&internal.ImportValueRequest{ + buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{ Index: index, Field: field, Shard: shard, @@ -685,7 +683,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, if uri == nil { panic("need to pass a URI to BlockData") } - buf, err := proto.Marshal(&internal.BlockDataRequest{ + buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{ Index: index, Field: field, Shard: shard, @@ -721,10 +719,10 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, } // Decode response object. - var rsp internal.BlockDataResponse + var rsp pilosa.BlockDataResponse if body, err := ioutil.ReadAll(resp.Body); err != nil { return nil, nil, errors.Wrap(err, "reading") - } else if err := proto.Unmarshal(body, &rsp); err != nil { + } else if err := c.serializer.Unmarshal(body, &rsp); err != nil { return nil, nil, errors.Wrap(err, "unmarshalling") } return rsp.RowIDs, rsp.ColumnIDs, nil diff --git a/http/handler.go b/http/handler.go index 33085879c..2b3d3ebd8 100644 --- a/http/handler.go +++ b/http/handler.go @@ -33,11 +33,9 @@ import ( "strings" "time" - "github.com/gogo/protobuf/proto" "github.com/gorilla/handlers" "github.com/gorilla/mux" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/internal" "github.com/pkg/errors" ) @@ -911,8 +909,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { if field.Type() == pilosa.FieldTypeInt { // Field type: Int // Marshal into request object. - var req internal.ImportValueRequest - if err := proto.Unmarshal(body, &req); err != nil { + req := &pilosa.ImportValueRequest{} + if err := h.API.Serializer.Unmarshal(body, req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -929,8 +927,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { } else { // Field type: Set, Time // Marshal into request object. - var req internal.ImportRequest - if err := proto.Unmarshal(body, &req); err != nil { + req := &pilosa.ImportRequest{} + if err := h.API.Serializer.Unmarshal(body, req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -947,7 +945,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { } // Marshal response object. - buf, e := proto.Marshal(&internal.ImportResponse{Err: ""}) + buf, e := h.API.Serializer.Marshal(&pilosa.ImportResponse{Err: ""}) if e != nil { http.Error(w, fmt.Sprintf("marshal import response"), http.StatusInternalServerError) return