scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove internal references from api and http/*

Browse source
This commit is contained in:
Matt Jaffee 2018-07-05 16:02:18 -05:00
parent bbb93abd57
commit db2a53223d
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
5 changed files with 197 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
api.go
View file

@ -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")
}

134
encoding/proto/proto.go
View file

@ -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)

37
handler.go
View file

@ -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
}

34
http/client.go
View file

@ -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

12
http/handler.go
View file

@ -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