From 83f3c398d96d43a9e05b32407a30edea0043245b Mon Sep 17 00:00:00 2001
From: Jan Semmelink <jan@uafrica.com>
Date: Mon, 29 Nov 2021 11:30:43 +0200
Subject: [PATCH] Add Get() method to time series
---
search/time_series.go | 65 ++++++++++++++++++++++++++++++++++++++++---
1 file changed, 61 insertions(+), 4 deletions(-)
diff --git a/search/time_series.go b/search/time_series.go
index 68f62e7..1928432 100644
--- a/search/time_series.go
+++ b/search/time_series.go
@@ -59,6 +59,7 @@ type timeSeries struct {
createdDates map[string]bool
searchResponseBodyType reflect.Type
+ getResponseBodyType reflect.Type
}
//purpose:
@@ -132,6 +133,18 @@ func (w *writer) TimeSeries(name string, tmpl interface{}) (TimeSeries, error) {
if err != nil {
return nil, errors.Wrapf(err, "failed to make search response type for time-series")
}
+
+ //define get response type
+ //similar to GetResponseBody
+ ts.getResponseBodyType, err = reflection.CloneType(
+ reflect.TypeOf(GetResponseBody{}),
+ map[string]reflect.Type{
+ "._source": ts.dataType,
+ })
+ if err != nil {
+ return nil, errors.Wrapf(err, "failed to make get response type for time-series")
+ }
+
w.timeSeriesByName[name] = ts
return ts, nil
}
@@ -452,12 +465,56 @@ func (ts *timeSeries) Search(query Query, limit int64) (docs map[string]interfac
docs = map[string]interface{}{}
for i := 0; i < hits.Len(); i++ {
hit := hits.Index(i)
- id := hit.Field(2).Interface().(string) //HitDoc.ID
- docs[id] = hit.Field(4).Interface() //HitDoc.Source
+ index := hit.Field(0).Interface().(string) //HitDoc.Index
+ id := hit.Field(2).Interface().(string) //HitDoc.ID
+ docs[index+"/"+id] = hit.Field(4).Interface() //HitDoc.Source
}
return docs, hitsTotalValue.Interface().(int), nil
}
-func (ts *timeSeries) Get(id string) (interface{}, error) {
- return nil, errors.Errorf("NYI")
+func (ds *timeSeries) Get(id string) (doc interface{}, err error) {
+ if ds == nil {
+ return nil, errors.Errorf("document store == nil")
+ }
+ get := opensearchapi.GetRequest{
+ Index: ds.name,
+ DocumentType: "_doc",
+ DocumentID: id,
+ }
+ getResponse, err := get.Do(context.Background(), ds.w.client)
+ if err != nil {
+ err = errors.Wrapf(err, "failed to get document")
+ return
+ }
+
+ switch getResponse.StatusCode {
+ case http.StatusOK:
+ default:
+ resBody, _ := ioutil.ReadAll(getResponse.Body)
+ err = errors.Errorf("Get failed with HTTP status %v: %s", getResponse.StatusCode, string(resBody))
+ return
+ }
+
+ resBodyPtrValue := reflect.New(ds.getResponseBodyType)
+ if err = json.NewDecoder(getResponse.Body).Decode(resBodyPtrValue.Interface()); err != nil {
+ err = errors.Wrapf(err, "cannot decode get response body")
+ return
+ }
+
+ foundVar, err := reflection.Get(resBodyPtrValue, ".found")
+ if err != nil {
+ err = errors.Wrapf(err, "cannot get found value")
+ return
+ }
+ if found, ok := foundVar.Interface().(bool); !ok || !found {
+ return nil, nil //not found
+ }
+
+ //found
+ source, err := reflection.Get(resBodyPtrValue, "._source")
+ if err != nil {
+ err = errors.Wrapf(err, "cannot get document from get response")
+ return
+ }
+ return source.Interface(), nil
}
--
GitLab