Convert RethinkDb dateTime structure in query-responses to RFC3339-fo…

…rmatted string

format/JsonFormatter.go

@@ -14,6 +14,7 @@
 package format
 
 import (
+	"encoding/json"
 	"fmt"
 	"io"
 	"reflect"
@@ -28,17 +29,16 @@ type jsonFormatter struct {
 
 // newJsonFormatter creates a new json formatter
 func newJsonFormatter(s *MarshalSpec) Formatter {
-	return &jsonFormatter{
-		MarshalSpec: s,
+	return &preFormatter{
+		&jsonFormatter{
+			MarshalSpec: s,
+		},
 	}
 }
 
 // WriteRecord writes a record to the formatters writer
 func (jf *jsonFormatter) WriteRecord(record interface{}) error {
 	switch v := record.(type) {
-	case string:
-		_, err := fmt.Fprint(jf.rWriter, v)
-		return err
 	case proto.Message:
 		var values reflect.Value
 		values = reflect.ValueOf(v).Elem().FieldByName("Value")
@@ -65,7 +65,12 @@ func (jf *jsonFormatter) WriteRecord(record interface{}) error {
 			}
 		}
 	default:
-		return fmt.Errorf("illegal record type '%T'", record)
+		j, err := json.Marshal(v)
+		if err != nil {
+			return err
+		}
+		_, err = fmt.Fprint(jf.rWriter, string(j))
+		return err
 	}
 	return nil
 }

format/TemplateFormatter.go

@@ -46,7 +46,7 @@ func newTemplateFormatter(s *MarshalSpec) (Formatter, error) {
 		return nil, err
 	}
 	t.parsedTemplate = pt
-	return t, nil
+	return &preFormatter{t}, nil
 }
 
 // WriteRecord writes a record to the formatters writer
@@ -108,16 +108,6 @@ func parseTemplate(templateString string) (*template.Template, error) {
 				return fmt.Sprintf("%-24.24s", ts.AsTime().Format(time.RFC3339))
 			}
 		},
-		"rethinktime": func(ts map[string]interface{}) string {
-			if ts == nil {
-				return "                        "
-			} else {
-				dateTime, _ := ts["dateTime"].(map[string]interface{})
-				date, _ := dateTime["date"].(map[string]interface{})
-				time, _ := dateTime["time"].(map[string]interface{})
-				return fmt.Sprintf("%04.f-%02.f-%02.fT%02.f:%02.f:%02.f", date["year"], date["month"], date["day"], time["hour"], time["minute"], time["second"])
-			}
-		},
 		"json": func(v interface{}) (string, error) {
 			if v == nil {
 				return "", nil

format/YamlFormatter.go

@@ -29,29 +29,16 @@ type yamlFormatter struct {
 
 // newYamlFormatter creates a new yaml formatter
 func newYamlFormatter(s *MarshalSpec) Formatter {
-	return &yamlFormatter{
-		MarshalSpec: s,
+	return &preFormatter{
+		&yamlFormatter{
+			MarshalSpec: s,
+		},
 	}
 }
 
 // WriteRecord writes a record to the formatters writer
 func (yf *yamlFormatter) WriteRecord(record interface{}) error {
 	switch v := record.(type) {
-	case string:
-		final, err := yaml.JSONToYAML([]byte(v))
-		if err != nil {
-			fmt.Printf("err: %v\n", err)
-			return err
-		}
-
-		_, err = fmt.Fprint(yf.rWriter, string(final))
-		if err != nil {
-			return err
-		}
-		_, err = fmt.Fprintln(yf.rWriter, "---")
-		if err != nil {
-			return err
-		}
 	case proto.Message:
 		var values reflect.Value
 		values = reflect.ValueOf(v).Elem().FieldByName("Value")
@@ -88,7 +75,20 @@ func (yf *yamlFormatter) WriteRecord(record interface{}) error {
 			return err
 		}
 	default:
-		return fmt.Errorf("illegal record type '%T'", record)
+		final, err := yaml.Marshal(record)
+		if err != nil {
+			fmt.Printf("err: %v\n", err)
+			return err
+		}
+
+		_, err = fmt.Fprint(yf.rWriter, string(final))
+		if err != nil {
+			return err
+		}
+		_, err = fmt.Fprintln(yf.rWriter, "---")
+		if err != nil {
+			return err
+		}
 	}
 	return nil
 }

format/formatter.go

@@ -21,15 +21,15 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
-	"os"
-	"strings"
-	"sync"
-
 	"github.com/invopop/yaml"
 	"github.com/nlnwa/veidemann-api/go/config/v1"
 	"github.com/rs/zerolog/log"
 	"google.golang.org/protobuf/encoding/protojson"
+	"io"
+	"os"
+	"strings"
+	"sync"
+	"time"
 )
 
 var jsonMarshaler = &protojson.MarshalOptions{EmitUnpopulated: true}
@@ -47,6 +47,132 @@ type Formatter interface {
 	Close() error
 }
 
+type anyRecord struct {
+	v interface{}
+}
+
+func (r *anyRecord) UnmarshalJSON(b []byte) error {
+	var i interface{}
+	err := json.Unmarshal(b, &i)
+	if err != nil {
+		return err
+	}
+
+	switch j := i.(type) {
+	case map[string]interface{}:
+		if d, ok := r.formatDate(j); ok {
+			r.v = d
+			return nil
+		}
+
+		r.traverseMap(&j)
+		r.v = j
+	default:
+		r.v = i
+	}
+
+	return err
+}
+
+func (r *anyRecord) traverseMap(i *map[string]interface{}) {
+	for k, v := range *i {
+		if m, ok := v.(map[string]interface{}); ok {
+			if d, ok := r.formatDate(m); ok {
+				(*i)[k] = d
+			} else {
+				r.traverseMap(&m)
+			}
+		}
+	}
+}
+
+// getAsInt returns the value as an int if it is a float64 or int
+func getAsInt(v interface{}) (int, bool) {
+	switch i := v.(type) {
+	case float64:
+		return int(i), true
+	case int:
+		return i, true
+	default:
+		return 0, false
+	}
+}
+
+// formatDate if i is recognized as a RethinkDb date, the date is returned as a RFC3339 formatted string
+func (r *anyRecord) formatDate(i map[string]interface{}) (string, bool) {
+	var year, month, day, hour, minute, second, nano, offset int
+
+	if dateTime, ok := i["dateTime"].(map[string]interface{}); !ok {
+		return "", false
+	} else {
+		if date, ok := dateTime["date"].(map[string]interface{}); !ok {
+			return "", false
+		} else {
+			if year, ok = getAsInt(date["year"]); !ok {
+				return "", false
+			}
+			if month, ok = getAsInt(date["month"]); !ok {
+				return "", false
+			}
+			if day, ok = getAsInt(date["day"]); !ok {
+				return "", false
+			}
+		}
+		if tm, ok := dateTime["time"].(map[string]interface{}); !ok {
+			return "", false
+		} else {
+			if hour, ok = getAsInt(tm["hour"]); !ok {
+				return "", false
+			}
+			if minute, ok = getAsInt(tm["minute"]); !ok {
+				return "", false
+			}
+			if second, ok = getAsInt(tm["second"]); !ok {
+				return "", false
+			}
+			if nano, ok = getAsInt(tm["nano"]); !ok {
+				return "", false
+			}
+		}
+	}
+	if of, ok := i["offset"].(map[string]interface{}); !ok {
+		return "", false
+	} else {
+		if offset, ok = getAsInt(of["totalSeconds"]); !ok {
+			return "", false
+		}
+	}
+	tz := time.UTC
+	if offset != 0 {
+		tz = time.FixedZone(fmt.Sprintf("OFF%.d", offset), offset)
+	}
+	d := time.Date(year, time.Month(month), day, hour, minute, second, nano, tz)
+	return d.Format(time.RFC3339Nano), true
+}
+
+// preFormatter wraps a formatter and converts json strings to objects
+type preFormatter struct {
+	formatter Formatter
+}
+
+func (p *preFormatter) WriteRecord(record interface{}) error {
+	switch v := record.(type) {
+	case string:
+		var j anyRecord
+		err := json.Unmarshal([]byte(v), &j)
+		if err != nil {
+			return fmt.Errorf("failed to parse json: %w", err)
+		}
+		record = j.v
+	}
+
+	return p.formatter.WriteRecord(record)
+}
+
+func (p *preFormatter) Close() error {
+	return p.formatter.Close()
+}
+
 // MarshalSpec is the specification for a formatter
 type MarshalSpec struct {
 	ObjectType          string