(ctx context.Context, req *scatterRequest)
| 290 | } |
| 291 | |
| 292 | func (g *GrpcClient) Query(ctx context.Context, |
| 293 | req *scatterRequest) (*bleve.SearchResult, error) { |
| 294 | scatterGatherReq := &pb.SearchRequest{ |
| 295 | IndexName: g.IndexName, |
| 296 | IndexUUID: g.IndexUUID, |
| 297 | } |
| 298 | |
| 299 | b, err := MarshalJSON(req.searchRequest) |
| 300 | if err != nil { |
| 301 | return nil, err |
| 302 | } |
| 303 | scatterGatherReq.Contents = b |
| 304 | |
| 305 | b, err = MarshalJSON(req.ctlParams) |
| 306 | if err != nil { |
| 307 | return nil, err |
| 308 | } |
| 309 | scatterGatherReq.QueryCtlParams = b |
| 310 | |
| 311 | b, err = MarshalJSON(req.onlyPIndexes) |
| 312 | if err != nil { |
| 313 | return nil, err |
| 314 | } |
| 315 | scatterGatherReq.QueryPIndexes = b |
| 316 | |
| 317 | // check if stream rpc is requested |
| 318 | if se := ctx.Value(search.MakeDocumentMatchHandlerKey); se != nil { |
| 319 | if _, ok := se.(search.MakeDocumentMatchHandler); ok { |
| 320 | scatterGatherReq.Stream = true |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | // mark that its a scatter gather query |
| 325 | nctx := metadata.AppendToOutgoingContext(ctx, |
| 326 | rpcClusterActionKey, clusterActionScatterGather) |
| 327 | |
| 328 | result, er := g.SearchRPC(nctx, req, scatterGatherReq) |
| 329 | if st, ok := status.FromError(er); ok { |
| 330 | g.lastSearchStatus = httpStatusCodes(st.Code()) |
| 331 | if g.lastSearchStatus == http.StatusOK { |
| 332 | return result, nil |
| 333 | } |
| 334 | g.lastErrBody, _ = MarshalJSON(err) |
| 335 | return nil, fmt.Errorf("grpc_client: query got status code: %d,"+ |
| 336 | " resp: %#v, err: %v", |
| 337 | g.lastSearchStatus, result, er) |
| 338 | } |
| 339 | |
| 340 | return result, fmt.Errorf("grpc_client: invalid status code, err: %v", er) |
| 341 | } |
| 342 | |
| 343 | func (g *GrpcClient) Advanced() (index.Index, error) { |
| 344 | return nil, indexClientUnimplementedErr |
no test coverage detected