Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 debug.go package eventstream import ( "bytes" "encoding/base64" "encoding/json" "fmt" "strconv" ) type decodedMessage struct { rawMessage Headers decodedHeaders `json:"headers"` } type jsonMessage struct { Length json.Number `json:"total_length"` HeadersLen json.Number `json:"headers_length"` PreludeCRC json.Number `json:"prelude_crc"` Headers decodedHeaders `json:"headers"` Payload []byte `json:"payload"` CRC json.Number `json:"message_crc"` } func (d decodedMessage) UnmarshalJSON(b []byte) (err error) { var jsonMsg jsonMessage if err = json.Unmarshal(b, &jsonMsg); err != nil { return err } d.Length, err = numAsUint32(jsonMsg.Length) if err != nil { return err } d.HeadersLen, err = numAsUint32(jsonMsg.HeadersLen) if err != nil { return err } d.PreludeCRC, err = numAsUint32(jsonMsg.PreludeCRC) if err != nil { return err } d.Headers = jsonMsg.Headers d.Payload = jsonMsg.Payload d.CRC, err = numAsUint32(jsonMsg.CRC) if err != nil { return err } return nil } func (d decodedMessage) MarshalJSON() ([]byte, error) { jsonMsg := jsonMessage{ Length: json.Number(strconv.Itoa(int(d.Length))), HeadersLen: json.Number(strconv.Itoa(int(d.HeadersLen))), PreludeCRC: json.Number(strconv.Itoa(int(d.PreludeCRC))), Headers: d.Headers, Payload: d.Payload, CRC: json.Number(strconv.Itoa(int(d.CRC))), } return json.Marshal(jsonMsg) } func numAsUint32(n json.Number) (uint32, error) { v, err := n.Int64() if err != nil { return 0, fmt.Errorf("failed to get int64 json number, %v", err) } return uint32(v), nil } func (d decodedMessage) Message() Message { return Message{ Headers: Headers(d.Headers), Payload: d.Payload, } } type decodedHeaders Headers func (hs decodedHeaders) UnmarshalJSON(b []byte) error { var jsonHeaders []struct { Name string `json:"name"` Type valueType `json:"type"` Value interface{} `json:"value"` } decoder := json.NewDecoder(bytes.NewReader(b)) decoder.UseNumber() if err := decoder.Decode(&jsonHeaders); err != nil { return err } var headers Headers for _, h := range jsonHeaders { value, err := valueFromType(h.Type, h.Value) if err != nil { return err } headers.Set(h.Name, value) } hs = decodedHeaders(headers) return nil } func valueFromType(typ valueType, val interface{}) (Value, error) { switch typ { case trueValueType: return BoolValue(true), nil case falseValueType: return BoolValue(false), nil case int8ValueType: v, err := val.(json.Number).Int64() return Int8Value(int8(v)), err case int16ValueType: v, err := val.(json.Number).Int64() return Int16Value(int16(v)), err case int32ValueType: v, err := val.(json.Number).Int64() return Int32Value(int32(v)), err case int64ValueType: v, err := val.(json.Number).Int64() return Int64Value(v), err case bytesValueType: v, err := base64.StdEncoding.DecodeString(val.(string)) return BytesValue(v), err case stringValueType: v, err := base64.StdEncoding.DecodeString(val.(string)) return StringValue(string(v)), err case timestampValueType: v, err := val.(json.Number).Int64() return TimestampValue(timeFromEpochMilli(v)), err case uuidValueType: v, err := base64.StdEncoding.DecodeString(val.(string)) var tv UUIDValue copy(tv[:], v) return tv, err default: panic(fmt.Sprintf("unknown type, %s, %T", typ.String(), val)) } }	3.36 KB	2025-05-25 12:27
📄 decode.go package eventstream import ( "bytes" "encoding/binary" "encoding/hex" "encoding/json" "fmt" "hash" "hash/crc32" "io" "github.com/aws/aws-sdk-go/aws" ) // Decoder provides decoding of an Event Stream messages. type Decoder struct { r io.Reader logger aws.Logger } // NewDecoder initializes and returns a Decoder for decoding event // stream messages from the reader provided. func NewDecoder(r io.Reader, opts ...func(Decoder)) Decoder { d := &Decoder{ r: r, } for _, opt := range opts { opt(d) } return d } // DecodeWithLogger adds a logger to be used by the decoder when decoding // stream events. func DecodeWithLogger(logger aws.Logger) func(Decoder) { return func(d Decoder) { d.logger = logger } } // Decode attempts to decode a single message from the event stream reader. // Will return the event stream message, or error if Decode fails to read // the message from the stream. func (d Decoder) Decode(payloadBuf []byte) (m Message, err error) { reader := d.r if d.logger != nil { debugMsgBuf := bytes.NewBuffer(nil) reader = io.TeeReader(reader, debugMsgBuf) defer func() { logMessageDecode(d.logger, debugMsgBuf, m, err) }() } m, err = Decode(reader, payloadBuf) return m, err } // Decode attempts to decode a single message from the event stream reader. // Will return the event stream message, or error if Decode fails to read // the message from the reader. func Decode(reader io.Reader, payloadBuf []byte) (m Message, err error) { crc := crc32.New(crc32IEEETable) hashReader := io.TeeReader(reader, crc) prelude, err := decodePrelude(hashReader, crc) if err != nil { return Message{}, err } if prelude.HeadersLen > 0 { lr := io.LimitReader(hashReader, int64(prelude.HeadersLen)) m.Headers, err = decodeHeaders(lr) if err != nil { return Message{}, err } } if payloadLen := prelude.PayloadLen(); payloadLen > 0 { buf, err := decodePayload(payloadBuf, io.LimitReader(hashReader, int64(payloadLen))) if err != nil { return Message{}, err } m.Payload = buf } msgCRC := crc.Sum32() if err := validateCRC(reader, msgCRC); err != nil { return Message{}, err } return m, nil } func logMessageDecode(logger aws.Logger, msgBuf bytes.Buffer, msg Message, decodeErr error) { w := bytes.NewBuffer(nil) defer func() { logger.Log(w.String()) }() fmt.Fprintf(w, "Raw message:\n%s\n", hex.Dump(msgBuf.Bytes())) if decodeErr != nil { fmt.Fprintf(w, "Decode error: %v\n", decodeErr) return } rawMsg, err := msg.rawMessage() if err != nil { fmt.Fprintf(w, "failed to create raw message, %v\n", err) return } decodedMsg := decodedMessage{ rawMessage: rawMsg, Headers: decodedHeaders(msg.Headers), } fmt.Fprintf(w, "Decoded message:\n") encoder := json.NewEncoder(w) if err := encoder.Encode(decodedMsg); err != nil { fmt.Fprintf(w, "failed to generate decoded message, %v\n", err) } } func decodePrelude(r io.Reader, crc hash.Hash32) (messagePrelude, error) { var p messagePrelude var err error p.Length, err = decodeUint32(r) if err != nil { return messagePrelude{}, err } p.HeadersLen, err = decodeUint32(r) if err != nil { return messagePrelude{}, err } if err := p.ValidateLens(); err != nil { return messagePrelude{}, err } preludeCRC := crc.Sum32() if err := validateCRC(r, preludeCRC); err != nil { return messagePrelude{}, err } p.PreludeCRC = preludeCRC return p, nil } func decodePayload(buf []byte, r io.Reader) ([]byte, error) { w := bytes.NewBuffer(buf[0:0]) _, err := io.Copy(w, r) return w.Bytes(), err } func decodeUint8(r io.Reader) (uint8, error) { type byteReader interface { ReadByte() (byte, error) } if br, ok := r.(byteReader); ok { v, err := br.ReadByte() return uint8(v), err } var b [1]byte _, err := io.ReadFull(r, b[:]) return uint8(b[0]), err } func decodeUint16(r io.Reader) (uint16, error) { var b [2]byte bs := b[:] _, err := io.ReadFull(r, bs) if err != nil { return 0, err } return binary.BigEndian.Uint16(bs), nil } func decodeUint32(r io.Reader) (uint32, error) { var b [4]byte bs := b[:] _, err := io.ReadFull(r, bs) if err != nil { return 0, err } return binary.BigEndian.Uint32(bs), nil } func decodeUint64(r io.Reader) (uint64, error) { var b [8]byte bs := b[:] _, err := io.ReadFull(r, bs) if err != nil { return 0, err } return binary.BigEndian.Uint64(bs), nil } func validateCRC(r io.Reader, expect uint32) error { msgCRC, err := decodeUint32(r) if err != nil { return err } if msgCRC != expect { return ChecksumError{} } return nil }	4.5 KB	2025-05-25 12:27
📄 decode_test.go package eventstream import ( "bytes" "encoding/hex" "io/ioutil" "os" "reflect" "testing" ) func TestWriteEncodedFromDecoded(t testing.T) { cases, err := readPositiveTests("testdata") if err != nil { t.Fatalf("failed to load positive tests, %v", err) } for _, c := range cases { f, err := ioutil.TempFile(os.TempDir(), "encoded_positive_"+c.Name) if err != nil { t.Fatalf("failed to open %q, %v", c.Name, err) } encoder := NewEncoder(f) msg := c.Decoded.Message() if err := encoder.Encode(msg); err != nil { t.Errorf("failed to encode %q, %v", c.Name, err) } if err = f.Close(); err != nil { t.Errorf("expected %v, got %v", "no error", err) } if err = os.Remove(f.Name()); err != nil { t.Errorf("expected %v, got %v", "no error", err) } } } func TestDecoder_Decode(t testing.T) { cases, err := readPositiveTests("testdata") if err != nil { t.Fatalf("failed to load positive tests, %v", err) } for _, c := range cases { decoder := NewDecoder(bytes.NewBuffer(c.Encoded)) msg, err := decoder.Decode(nil) if err != nil { t.Fatalf("%s, expect no decode error, got %v", c.Name, err) } raw, err := msg.rawMessage() // rawMessage will fail if payload read CRC fails if err != nil { t.Fatalf("%s, failed to get raw decoded message %v", c.Name, err) } if e, a := c.Decoded.Length, raw.Length; e != a { t.Errorf("%s, expect %v length, got %v", c.Name, e, a) } if e, a := c.Decoded.HeadersLen, raw.HeadersLen; e != a { t.Errorf("%s, expect %v HeadersLen, got %v", c.Name, e, a) } if e, a := c.Decoded.PreludeCRC, raw.PreludeCRC; e != a { t.Errorf("%s, expect %v PreludeCRC, got %v", c.Name, e, a) } if e, a := Headers(c.Decoded.Headers), msg.Headers; !reflect.DeepEqual(e, a) { t.Errorf("%s, expect %v headers, got %v", c.Name, e, a) } if e, a := c.Decoded.Payload, raw.Payload; !bytes.Equal(e, a) { t.Errorf("%s, expect %v payload, got %v", c.Name, e, a) } if e, a := c.Decoded.CRC, raw.CRC; e != a { t.Errorf("%s, expect %v CRC, got %v", c.Name, e, a) } } } func TestDecoder_Decode_Negative(t testing.T) { cases, err := readNegativeTests("testdata") if err != nil { t.Fatalf("failed to load negative tests, %v", err) } for _, c := range cases { decoder := NewDecoder(bytes.NewBuffer(c.Encoded)) msg, err := decoder.Decode(nil) if err == nil { rawMsg, rawMsgErr := msg.rawMessage() t.Fatalf("%s, expect error, got none, %s,\n%s\n%#v, %v\n", c.Name, c.Err, hex.Dump(c.Encoded), rawMsg, rawMsgErr) } } } var testEncodedMsg = []byte{0, 0, 0, 61, 0, 0, 0, 32, 7, 253, 131, 150, 12, 99, 111, 110, 116, 101, 110, 116, 45, 116, 121, 112, 101, 7, 0, 16, 97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 47, 106, 115, 111, 110, 123, 39, 102, 111, 111, 39, 58, 39, 98, 97, 114, 39, 125, 141, 156, 8, 177} func TestDecoder_DecodeMultipleMessages(t testing.T) { const ( expectMsgCount = 10 expectPayloadLen = 13 ) r := bytes.NewBuffer(nil) for i := 0; i < expectMsgCount; i++ { r.Write(testEncodedMsg) } decoder := NewDecoder(r) var err error var msg Message var count int for { msg, err = decoder.Decode(nil) if err != nil { break } count++ if e, a := expectPayloadLen, len(msg.Payload); e != a { t.Errorf("expect %v payload len, got %v", e, a) } if e, a := []byte(`{'foo':'bar'}`), msg.Payload; !bytes.Equal(e, a) { t.Errorf("expect %v payload, got %v", e, a) } } type causer interface { Cause() error } if err != nil && count != expectMsgCount { t.Fatalf("expect, no error, got %v", err) } if e, a := expectMsgCount, count; e != a { t.Errorf("expect %v messages read, got %v", e, a) } } func BenchmarkDecode(b testing.B) { r := bytes.NewReader(testEncodedMsg) decoder := NewDecoder(r) payloadBuf := make([]byte, 0, 51024) b.ResetTimer() for i := 0; i < b.N; i++ { msg, err := decoder.Decode(payloadBuf) if err != nil { b.Fatal(err) } // Release the payload buffer payloadBuf = msg.Payload[0:0] r.Seek(0, 0) } } func BenchmarkDecode_NoPayloadBuf(b *testing.B) { r := bytes.NewReader(testEncodedMsg) decoder := NewDecoder(r) b.ResetTimer() for i := 0; i < b.N; i++ { _, err := decoder.Decode(nil) if err != nil { b.Fatal(err) } r.Seek(0, 0) } }	4.19 KB	2025-05-25 12:27
📄 encode.go package eventstream import ( "bytes" "encoding/binary" "encoding/hex" "encoding/json" "fmt" "hash" "hash/crc32" "io" "github.com/aws/aws-sdk-go/aws" ) // Encoder provides EventStream message encoding. type Encoder struct { w io.Writer logger aws.Logger headersBuf bytes.Buffer } // NewEncoder initializes and returns an Encoder to encode Event Stream // messages to an io.Writer. func NewEncoder(w io.Writer, opts ...func(Encoder)) Encoder { e := &Encoder{ w: w, headersBuf: bytes.NewBuffer(nil), } for _, opt := range opts { opt(e) } return e } // EncodeWithLogger adds a logger to be used by the encode when decoding // stream events. func EncodeWithLogger(logger aws.Logger) func(Encoder) { return func(d Encoder) { d.logger = logger } } // Encode encodes a single EventStream message to the io.Writer the Encoder // was created with. An error is returned if writing the message fails. func (e Encoder) Encode(msg Message) (err error) { e.headersBuf.Reset() writer := e.w if e.logger != nil { encodeMsgBuf := bytes.NewBuffer(nil) writer = io.MultiWriter(writer, encodeMsgBuf) defer func() { logMessageEncode(e.logger, encodeMsgBuf, msg, err) }() } if err = EncodeHeaders(e.headersBuf, msg.Headers); err != nil { return err } crc := crc32.New(crc32IEEETable) hashWriter := io.MultiWriter(writer, crc) headersLen := uint32(e.headersBuf.Len()) payloadLen := uint32(len(msg.Payload)) if err = encodePrelude(hashWriter, crc, headersLen, payloadLen); err != nil { return err } if headersLen > 0 { if _, err = io.Copy(hashWriter, e.headersBuf); err != nil { return err } } if payloadLen > 0 { if _, err = hashWriter.Write(msg.Payload); err != nil { return err } } msgCRC := crc.Sum32() return binary.Write(writer, binary.BigEndian, msgCRC) } func logMessageEncode(logger aws.Logger, msgBuf *bytes.Buffer, msg Message, encodeErr error) { w := bytes.NewBuffer(nil) defer func() { logger.Log(w.String()) }() fmt.Fprintf(w, "Message to encode:\n") encoder := json.NewEncoder(w) if err := encoder.Encode(msg); err != nil { fmt.Fprintf(w, "Failed to get encoded message, %v\n", err) } if encodeErr != nil { fmt.Fprintf(w, "Encode error: %v\n", encodeErr) return } fmt.Fprintf(w, "Raw message:\n%s\n", hex.Dump(msgBuf.Bytes())) } func encodePrelude(w io.Writer, crc hash.Hash32, headersLen, payloadLen uint32) error { p := messagePrelude{ Length: minMsgLen + headersLen + payloadLen, HeadersLen: headersLen, } if err := p.ValidateLens(); err != nil { return err } err := binaryWriteFields(w, binary.BigEndian, p.Length, p.HeadersLen, ) if err != nil { return err } p.PreludeCRC = crc.Sum32() err = binary.Write(w, binary.BigEndian, p.PreludeCRC) if err != nil { return err } return nil } // EncodeHeaders writes the header values to the writer encoded in the event // stream format. Returns an error if a header fails to encode. func EncodeHeaders(w io.Writer, headers Headers) error { for _, h := range headers { hn := headerName{ Len: uint8(len(h.Name)), } copy(hn.Name[:hn.Len], h.Name) if err := hn.encode(w); err != nil { return err } if err := h.Value.encode(w); err != nil { return err } } return nil } func binaryWriteFields(w io.Writer, order binary.ByteOrder, vs ...interface{}) error { for _, v := range vs { if err := binary.Write(w, order, v); err != nil { return err } } return nil }	3.41 KB	2025-05-25 12:27
📄 encode_test.go package eventstream import ( "bytes" "encoding/hex" "reflect" "testing" ) func TestEncoder_Encode(t testing.T) { cases, err := readPositiveTests("testdata") if err != nil { t.Fatalf("failed to load positive tests, %v", err) } for _, c := range cases { var w bytes.Buffer encoder := NewEncoder(&w) err = encoder.Encode(c.Decoded.Message()) if err != nil { t.Fatalf("%s, failed to encode message, %v", c.Name, err) } if e, a := c.Encoded, w.Bytes(); !reflect.DeepEqual(e, a) { t.Errorf("%s, expect:\n%v\nactual:\n%v\n", c.Name, hex.Dump(e), hex.Dump(a)) } } } func BenchmarkEncode(b testing.B) { var w bytes.Buffer encoder := NewEncoder(&w) msg := Message{ Headers: Headers{ {Name: "event-id", Value: Int16Value(123)}, }, Payload: []byte(`{"abc":123}`), } b.ResetTimer() for i := 0; i < b.N; i++ { err := encoder.Encode(msg) if err != nil { b.Fatal(err) } } }	928 B	2025-05-25 12:27
📄 error.go package eventstream import "fmt" // LengthError provides the error for items being larger than a maximum length. type LengthError struct { Part string Want int Have int Value interface{} } func (e LengthError) Error() string { return fmt.Sprintf("%s length invalid, %d/%d, %v", e.Part, e.Want, e.Have, e.Value) } // ChecksumError provides the error for message checksum invalidation errors. type ChecksumError struct{} func (e ChecksumError) Error() string { return "message checksum mismatch" }	512 B	2025-05-25 12:27
📁 eventstreamapi	-	2025-05-25 12:27
📁 eventstreamtest	-	2025-05-25 12:27
📄 header.go package eventstream import ( "encoding/binary" "fmt" "io" ) // Headers are a collection of EventStream header values. type Headers []Header // Header is a single EventStream Key Value header pair. type Header struct { Name string Value Value } // Set associates the name with a value. If the header name already exists in // the Headers the value will be replaced with the new one. func (hs Headers) Set(name string, value Value) { var i int for ; i < len(hs); i++ { if (hs)[i].Name == name { (hs)[i].Value = value return } } hs = append(hs, Header{ Name: name, Value: value, }) } // Get returns the Value associated with the header. Nil is returned if the // value does not exist. func (hs Headers) Get(name string) Value { for i := 0; i < len(hs); i++ { if h := hs[i]; h.Name == name { return h.Value } } return nil } // Del deletes the value in the Headers if it exists. func (hs Headers) Del(name string) { for i := 0; i < len(hs); i++ { if (hs)[i].Name == name { copy((hs)[i:], (hs)[i+1:]) (hs) = (hs)[:len(hs)-1] } } } // Clone returns a deep copy of the headers func (hs Headers) Clone() Headers { o := make(Headers, 0, len(hs)) for _, h := range hs { o.Set(h.Name, h.Value) } return o } func decodeHeaders(r io.Reader) (Headers, error) { hs := Headers{} for { name, err := decodeHeaderName(r) if err != nil { if err == io.EOF { // EOF while getting header name means no more headers break } return nil, err } value, err := decodeHeaderValue(r) if err != nil { return nil, err } hs.Set(name, value) } return hs, nil } func decodeHeaderName(r io.Reader) (string, error) { var n headerName var err error n.Len, err = decodeUint8(r) if err != nil { return "", err } name := n.Name[:n.Len] if _, err := io.ReadFull(r, name); err != nil { return "", err } return string(name), nil } func decodeHeaderValue(r io.Reader) (Value, error) { var raw rawValue typ, err := decodeUint8(r) if err != nil { return nil, err } raw.Type = valueType(typ) var v Value switch raw.Type { case trueValueType: v = BoolValue(true) case falseValueType: v = BoolValue(false) case int8ValueType: var tv Int8Value err = tv.decode(r) v = tv case int16ValueType: var tv Int16Value err = tv.decode(r) v = tv case int32ValueType: var tv Int32Value err = tv.decode(r) v = tv case int64ValueType: var tv Int64Value err = tv.decode(r) v = tv case bytesValueType: var tv BytesValue err = tv.decode(r) v = tv case stringValueType: var tv StringValue err = tv.decode(r) v = tv case timestampValueType: var tv TimestampValue err = tv.decode(r) v = tv case uuidValueType: var tv UUIDValue err = tv.decode(r) v = tv default: panic(fmt.Sprintf("unknown value type %d", raw.Type)) } // Error could be EOF, let caller deal with it return v, err } const maxHeaderNameLen = 255 type headerName struct { Len uint8 Name [maxHeaderNameLen]byte } func (v headerName) encode(w io.Writer) error { if err := binary.Write(w, binary.BigEndian, v.Len); err != nil { return err } _, err := w.Write(v.Name[:v.Len]) return err }	3.13 KB	2025-05-25 12:27
📄 header_test.go package eventstream import ( "reflect" "testing" "time" ) func TestHeaders_Set(t testing.T) { expect := Headers{ {Name: "ABC", Value: StringValue("123")}, {Name: "EFG", Value: TimestampValue(time.Time{})}, } var actual Headers actual.Set("ABC", Int32Value(123)) actual.Set("ABC", StringValue("123")) // replace case actual.Set("EFG", TimestampValue(time.Time{})) if e, a := expect, actual; !reflect.DeepEqual(e, a) { t.Errorf("expect %v headers, got %v", e, a) } } func TestHeaders_Get(t testing.T) { headers := Headers{ {Name: "ABC", Value: StringValue("123")}, {Name: "EFG", Value: TimestampValue(time.Time{})}, } cases := []struct { Name string Value Value }{ {Name: "ABC", Value: StringValue("123")}, {Name: "EFG", Value: TimestampValue(time.Time{})}, {Name: "NotFound"}, } for i, c := range cases { actual := headers.Get(c.Name) if e, a := c.Value, actual; !reflect.DeepEqual(e, a) { t.Errorf("%d, expect %v value, got %v", i, e, a) } } } func TestHeaders_Del(t *testing.T) { headers := Headers{ {Name: "ABC", Value: StringValue("123")}, {Name: "EFG", Value: TimestampValue(time.Time{})}, {Name: "HIJ", Value: StringValue("123")}, {Name: "KML", Value: TimestampValue(time.Time{})}, } expectAfterDel := Headers{ {Name: "EFG", Value: TimestampValue(time.Time{})}, } headers.Del("HIJ") headers.Del("ABC") headers.Del("KML") if e, a := expectAfterDel, headers; !reflect.DeepEqual(e, a) { t.Errorf("expect %v headers, got %v", e, a) } }	1.48 KB	2025-05-25 12:27
📄 header_value.go package eventstream import ( "encoding/base64" "encoding/binary" "fmt" "io" "strconv" "time" ) const maxHeaderValueLen = 1<<15 - 1 // 2^15-1 or 32KB - 1 // valueType is the EventStream header value type. type valueType uint8 // Header value types const ( trueValueType valueType = iota falseValueType int8ValueType // Byte int16ValueType // Short int32ValueType // Integer int64ValueType // Long bytesValueType stringValueType timestampValueType uuidValueType ) func (t valueType) String() string { switch t { case trueValueType: return "bool" case falseValueType: return "bool" case int8ValueType: return "int8" case int16ValueType: return "int16" case int32ValueType: return "int32" case int64ValueType: return "int64" case bytesValueType: return "byte_array" case stringValueType: return "string" case timestampValueType: return "timestamp" case uuidValueType: return "uuid" default: return fmt.Sprintf("unknown value type %d", uint8(t)) } } type rawValue struct { Type valueType Len uint16 // Only set for variable length slices Value []byte // byte representation of value, BigEndian encoding. } func (r rawValue) encodeScalar(w io.Writer, v interface{}) error { return binaryWriteFields(w, binary.BigEndian, r.Type, v, ) } func (r rawValue) encodeFixedSlice(w io.Writer, v []byte) error { binary.Write(w, binary.BigEndian, r.Type) _, err := w.Write(v) return err } func (r rawValue) encodeBytes(w io.Writer, v []byte) error { if len(v) > maxHeaderValueLen { return LengthError{ Part: "header value", Want: maxHeaderValueLen, Have: len(v), Value: v, } } r.Len = uint16(len(v)) err := binaryWriteFields(w, binary.BigEndian, r.Type, r.Len, ) if err != nil { return err } _, err = w.Write(v) return err } func (r rawValue) encodeString(w io.Writer, v string) error { if len(v) > maxHeaderValueLen { return LengthError{ Part: "header value", Want: maxHeaderValueLen, Have: len(v), Value: v, } } r.Len = uint16(len(v)) type stringWriter interface { WriteString(string) (int, error) } err := binaryWriteFields(w, binary.BigEndian, r.Type, r.Len, ) if err != nil { return err } if sw, ok := w.(stringWriter); ok { _, err = sw.WriteString(v) } else { _, err = w.Write([]byte(v)) } return err } func decodeFixedBytesValue(r io.Reader, buf []byte) error { _, err := io.ReadFull(r, buf) return err } func decodeBytesValue(r io.Reader) ([]byte, error) { var raw rawValue var err error raw.Len, err = decodeUint16(r) if err != nil { return nil, err } buf := make([]byte, raw.Len) _, err = io.ReadFull(r, buf) if err != nil { return nil, err } return buf, nil } func decodeStringValue(r io.Reader) (string, error) { v, err := decodeBytesValue(r) return string(v), err } // Value represents the abstract header value. type Value interface { Get() interface{} String() string valueType() valueType encode(io.Writer) error } // An BoolValue provides eventstream encoding, and representation // of a Go bool value. type BoolValue bool // Get returns the underlying type func (v BoolValue) Get() interface{} { return bool(v) } // valueType returns the EventStream header value type value. func (v BoolValue) valueType() valueType { if v { return trueValueType } return falseValueType } func (v BoolValue) String() string { return strconv.FormatBool(bool(v)) } // encode encodes the BoolValue into an eventstream binary value // representation. func (v BoolValue) encode(w io.Writer) error { return binary.Write(w, binary.BigEndian, v.valueType()) } // An Int8Value provides eventstream encoding, and representation of a Go // int8 value. type Int8Value int8 // Get returns the underlying value. func (v Int8Value) Get() interface{} { return int8(v) } // valueType returns the EventStream header value type value. func (Int8Value) valueType() valueType { return int8ValueType } func (v Int8Value) String() string { return fmt.Sprintf("0x%02x", int8(v)) } // encode encodes the Int8Value into an eventstream binary value // representation. func (v Int8Value) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeScalar(w, v) } func (v Int8Value) decode(r io.Reader) error { n, err := decodeUint8(r) if err != nil { return err } v = Int8Value(n) return nil } // An Int16Value provides eventstream encoding, and representation of a Go // int16 value. type Int16Value int16 // Get returns the underlying value. func (v Int16Value) Get() interface{} { return int16(v) } // valueType returns the EventStream header value type value. func (Int16Value) valueType() valueType { return int16ValueType } func (v Int16Value) String() string { return fmt.Sprintf("0x%04x", int16(v)) } // encode encodes the Int16Value into an eventstream binary value // representation. func (v Int16Value) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeScalar(w, v) } func (v Int16Value) decode(r io.Reader) error { n, err := decodeUint16(r) if err != nil { return err } v = Int16Value(n) return nil } // An Int32Value provides eventstream encoding, and representation of a Go // int32 value. type Int32Value int32 // Get returns the underlying value. func (v Int32Value) Get() interface{} { return int32(v) } // valueType returns the EventStream header value type value. func (Int32Value) valueType() valueType { return int32ValueType } func (v Int32Value) String() string { return fmt.Sprintf("0x%08x", int32(v)) } // encode encodes the Int32Value into an eventstream binary value // representation. func (v Int32Value) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeScalar(w, v) } func (v Int32Value) decode(r io.Reader) error { n, err := decodeUint32(r) if err != nil { return err } v = Int32Value(n) return nil } // An Int64Value provides eventstream encoding, and representation of a Go // int64 value. type Int64Value int64 // Get returns the underlying value. func (v Int64Value) Get() interface{} { return int64(v) } // valueType returns the EventStream header value type value. func (Int64Value) valueType() valueType { return int64ValueType } func (v Int64Value) String() string { return fmt.Sprintf("0x%016x", int64(v)) } // encode encodes the Int64Value into an eventstream binary value // representation. func (v Int64Value) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeScalar(w, v) } func (v Int64Value) decode(r io.Reader) error { n, err := decodeUint64(r) if err != nil { return err } v = Int64Value(n) return nil } // An BytesValue provides eventstream encoding, and representation of a Go // byte slice. type BytesValue []byte // Get returns the underlying value. func (v BytesValue) Get() interface{} { return []byte(v) } // valueType returns the EventStream header value type value. func (BytesValue) valueType() valueType { return bytesValueType } func (v BytesValue) String() string { return base64.StdEncoding.EncodeToString([]byte(v)) } // encode encodes the BytesValue into an eventstream binary value // representation. func (v BytesValue) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeBytes(w, []byte(v)) } func (v BytesValue) decode(r io.Reader) error { buf, err := decodeBytesValue(r) if err != nil { return err } v = BytesValue(buf) return nil } // An StringValue provides eventstream encoding, and representation of a Go // string. type StringValue string // Get returns the underlying value. func (v StringValue) Get() interface{} { return string(v) } // valueType returns the EventStream header value type value. func (StringValue) valueType() valueType { return stringValueType } func (v StringValue) String() string { return string(v) } // encode encodes the StringValue into an eventstream binary value // representation. func (v StringValue) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeString(w, string(v)) } func (v StringValue) decode(r io.Reader) error { s, err := decodeStringValue(r) if err != nil { return err } v = StringValue(s) return nil } // An TimestampValue provides eventstream encoding, and representation of a Go // timestamp. type TimestampValue time.Time // Get returns the underlying value. func (v TimestampValue) Get() interface{} { return time.Time(v) } // valueType returns the EventStream header value type value. func (TimestampValue) valueType() valueType { return timestampValueType } func (v TimestampValue) epochMilli() int64 { nano := time.Time(v).UnixNano() msec := nano / int64(time.Millisecond) return msec } func (v TimestampValue) String() string { msec := v.epochMilli() return strconv.FormatInt(msec, 10) } // encode encodes the TimestampValue into an eventstream binary value // representation. func (v TimestampValue) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } msec := v.epochMilli() return raw.encodeScalar(w, msec) } func (v TimestampValue) decode(r io.Reader) error { n, err := decodeUint64(r) if err != nil { return err } v = TimestampValue(timeFromEpochMilli(int64(n))) return nil } // MarshalJSON implements the json.Marshaler interface func (v TimestampValue) MarshalJSON() ([]byte, error) { return []byte(v.String()), nil } func timeFromEpochMilli(t int64) time.Time { secs := t / 1e3 msec := t % 1e3 return time.Unix(secs, msecint64(time.Millisecond)).UTC() } // An UUIDValue provides eventstream encoding, and representation of a UUID // value. type UUIDValue [16]byte // Get returns the underlying value. func (v UUIDValue) Get() interface{} { return v[:] } // valueType returns the EventStream header value type value. func (UUIDValue) valueType() valueType { return uuidValueType } func (v UUIDValue) String() string { return fmt.Sprintf(`%X-%X-%X-%X-%X`, v[0:4], v[4:6], v[6:8], v[8:10], v[10:]) } // encode encodes the UUIDValue into an eventstream binary value // representation. func (v UUIDValue) encode(w io.Writer) error { raw := rawValue{ Type: v.valueType(), } return raw.encodeFixedSlice(w, v[:]) } func (v UUIDValue) decode(r io.Reader) error { tv := (*v)[:] return decodeFixedBytesValue(r, tv) }	10.16 KB	2025-05-25 12:27
📄 header_value_test.go package eventstream import ( "bytes" "encoding/binary" "io" "reflect" "testing" "time" ) func binWrite(v interface{}) []byte { var w bytes.Buffer binary.Write(&w, binary.BigEndian, v) return w.Bytes() } var testValueEncodingCases = []struct { Val Value Expect []byte Decode func(io.Reader) (Value, error) }{ { BoolValue(true), []byte{byte(trueValueType)}, nil, }, { BoolValue(false), []byte{byte(falseValueType)}, nil, }, { Int8Value(0x0f), []byte{byte(int8ValueType), 0x0f}, func(r io.Reader) (Value, error) { var v Int8Value err := v.decode(r) return v, err }, }, { Int16Value(0x0f), append([]byte{byte(int16ValueType)}, binWrite(int16(0x0f))...), func(r io.Reader) (Value, error) { var v Int16Value err := v.decode(r) return v, err }, }, { Int32Value(0x0f), append([]byte{byte(int32ValueType)}, binWrite(int32(0x0f))...), func(r io.Reader) (Value, error) { var v Int32Value err := v.decode(r) return v, err }, }, { Int64Value(0x0f), append([]byte{byte(int64ValueType)}, binWrite(int64(0x0f))...), func(r io.Reader) (Value, error) { var v Int64Value err := v.decode(r) return v, err }, }, { BytesValue([]byte{0, 1, 2, 3}), []byte{byte(bytesValueType), 0x00, 0x04, 0, 1, 2, 3}, func(r io.Reader) (Value, error) { var v BytesValue err := v.decode(r) return v, err }, }, { StringValue("abc123"), append([]byte{byte(stringValueType), 0, 6}, []byte("abc123")...), func(r io.Reader) (Value, error) { var v StringValue err := v.decode(r) return v, err }, }, { TimestampValue( time.Date(2014, 04, 04, 0, 1, 0, 0, time.FixedZone("PDT", -7)), ), append([]byte{byte(timestampValueType)}, binWrite(int64(1396569667000))...), func(r io.Reader) (Value, error) { var v TimestampValue err := v.decode(r) return v, err }, }, { UUIDValue( [16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf}, ), []byte{byte(uuidValueType), 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf}, func(r io.Reader) (Value, error) { var v UUIDValue err := v.decode(r) return v, err }, }, } func TestValue_MarshalValue(t testing.T) { for i, c := range testValueEncodingCases { var w bytes.Buffer if err := c.Val.encode(&w); err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } if e, a := c.Expect, w.Bytes(); !reflect.DeepEqual(e, a) { t.Errorf("%d, expect %v, got %v", i, e, a) } } } func TestHeader_DecodeValues(t testing.T) { for i, c := range testValueEncodingCases { r := bytes.NewBuffer(c.Expect) v, err := decodeHeaderValue(r) if err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } switch tv := v.(type) { case TimestampValue: exp := time.Time(c.Val.(TimestampValue)) if e, a := exp, time.Time(tv); !e.Equal(a) { t.Errorf("%d, expect %v, got %v", i, e, a) } default: if e, a := c.Val, v; !reflect.DeepEqual(e, a) { t.Errorf("%d, expect %v, got %v", i, e, a) } } } } func TestValue_Decode(t testing.T) { for i, c := range testValueEncodingCases { if c.Decode == nil { continue } r := bytes.NewBuffer(c.Expect) r.ReadByte() // strip off Type field v, err := c.Decode(r) if err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } switch tv := v.(type) { case TimestampValue: exp := time.Time(c.Val.(TimestampValue)) if e, a := exp, time.Time(tv); !e.Equal(a) { t.Errorf("%d, expect %v, got %v", i, e, a) } default: if e, a := c.Val, v; !reflect.DeepEqual(e, a) { t.Errorf("%d, expect %v, got %v", i, e, a) } } } } func TestValue_String(t testing.T) { cases := []struct { Val Value Expect string }{ {BoolValue(true), "true"}, {BoolValue(false), "false"}, {Int8Value(0x0f), "0x0f"}, {Int16Value(0x0f), "0x000f"}, {Int32Value(0x0f), "0x0000000f"}, {Int64Value(0x0f), "0x000000000000000f"}, {BytesValue([]byte{0, 1, 2, 3}), "AAECAw=="}, {StringValue("abc123"), "abc123"}, {TimestampValue( time.Date(2014, 04, 04, 0, 1, 0, 0, time.FixedZone("PDT", -7)), ), "1396569667000", }, {UUIDValue([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe, 0xf}), "00010203-0405-0607-0809-0A0B0C0D0E0F", }, } for i, c := range cases { if e, a := c.Expect, c.Val.String(); e != a { t.Errorf("%d, expect %v, got %v", i, e, a) } } }	4.31 KB	2025-05-25 12:27
📄 message.go package eventstream import ( "bytes" "encoding/binary" "hash/crc32" ) const preludeLen = 8 const preludeCRCLen = 4 const msgCRCLen = 4 const minMsgLen = preludeLen + preludeCRCLen + msgCRCLen const maxPayloadLen = 1024 * 1024 * 16 // 16MB const maxHeadersLen = 1024 * 128 // 128KB const maxMsgLen = minMsgLen + maxHeadersLen + maxPayloadLen var crc32IEEETable = crc32.MakeTable(crc32.IEEE) // A Message provides the eventstream message representation. type Message struct { Headers Headers Payload []byte } func (m *Message) rawMessage() (rawMessage, error) { var raw rawMessage if len(m.Headers) > 0 { var headers bytes.Buffer if err := EncodeHeaders(&headers, m.Headers); err != nil { return rawMessage{}, err } raw.Headers = headers.Bytes() raw.HeadersLen = uint32(len(raw.Headers)) } raw.Length = raw.HeadersLen + uint32(len(m.Payload)) + minMsgLen hash := crc32.New(crc32IEEETable) binaryWriteFields(hash, binary.BigEndian, raw.Length, raw.HeadersLen) raw.PreludeCRC = hash.Sum32() binaryWriteFields(hash, binary.BigEndian, raw.PreludeCRC) if raw.HeadersLen > 0 { hash.Write(raw.Headers) } // Read payload bytes and update hash for it as well. if len(m.Payload) > 0 { raw.Payload = m.Payload hash.Write(raw.Payload) } raw.CRC = hash.Sum32() return raw, nil } // Clone returns a deep copy of the message. func (m Message) Clone() Message { var payload []byte if m.Payload != nil { payload = make([]byte, len(m.Payload)) copy(payload, m.Payload) } return Message{ Headers: m.Headers.Clone(), Payload: payload, } } type messagePrelude struct { Length uint32 HeadersLen uint32 PreludeCRC uint32 } func (p messagePrelude) PayloadLen() uint32 { return p.Length - p.HeadersLen - minMsgLen } func (p messagePrelude) ValidateLens() error { if p.Length == 0 \|\| p.Length > maxMsgLen { return LengthError{ Part: "message prelude", Want: maxMsgLen, Have: int(p.Length), } } if p.HeadersLen > maxHeadersLen { return LengthError{ Part: "message headers", Want: maxHeadersLen, Have: int(p.HeadersLen), } } if payloadLen := p.PayloadLen(); payloadLen > maxPayloadLen { return LengthError{ Part: "message payload", Want: maxPayloadLen, Have: int(payloadLen), } } return nil } type rawMessage struct { messagePrelude Headers []byte Payload []byte CRC uint32 }	2.33 KB	2025-05-25 12:27
📄 shared_test.go package eventstream import ( "bufio" "bytes" "encoding/json" "fmt" "io/ioutil" "path/filepath" "testing" ) type testCase struct { Name string Encoded []byte Decoded decodedMessage } type testErrorCase struct { Name string Encoded []byte Err string } type rawTestCase struct { Name string Encoded, Decoded []byte } func readRawTestCases(root, class string) (map[string]rawTestCase, error) { encoded, err := readTests(filepath.Join(root, "encoded", class)) if err != nil { return nil, err } decoded, err := readTests(filepath.Join(root, "decoded", class)) if err != nil { return nil, err } if len(encoded) == 0 { return nil, fmt.Errorf("expect encoded cases, found none") } if len(encoded) != len(decoded) { return nil, fmt.Errorf("encoded and decoded sets different") } rawCases := map[string]rawTestCase{} for name, encData := range encoded { decData, ok := decoded[name] if !ok { return nil, fmt.Errorf("encoded %q case not found in decoded set", name) } rawCases[name] = rawTestCase{ Name: name, Encoded: encData, Decoded: decData, } } return rawCases, nil } func readNegativeTests(root string) ([]testErrorCase, error) { rawCases, err := readRawTestCases(root, "negative") if err != nil { return nil, err } cases := make([]testErrorCase, 0, len(rawCases)) for name, rawCase := range rawCases { cases = append(cases, testErrorCase{ Name: name, Encoded: rawCase.Encoded, Err: string(rawCase.Decoded), }) } return cases, nil } func readPositiveTests(root string) ([]testCase, error) { rawCases, err := readRawTestCases(root, "positive") if err != nil { return nil, err } cases := make([]testCase, 0, len(rawCases)) for name, rawCase := range rawCases { var dec decodedMessage if err := json.Unmarshal(rawCase.Decoded, &dec); err != nil { return nil, fmt.Errorf("failed to decode %q, %v", name, err) } cases = append(cases, testCase{ Name: name, Encoded: rawCase.Encoded, Decoded: dec, }) } return cases, nil } func readTests(root string) (map[string][]byte, error) { items, err := ioutil.ReadDir(root) if err != nil { return nil, fmt.Errorf("failed to read test suite %q dirs, %v", root, err) } cases := map[string][]byte{} for _, item := range items { if item.IsDir() { continue } filename := filepath.Join(root, item.Name()) data, err := ioutil.ReadFile(filename) if err != nil { return nil, fmt.Errorf("failed to read test_data file %q, %v", filename, err) } cases[item.Name()] = data } return cases, nil } func compareLines(t *testing.T, a, b []byte) bool { as := bufio.NewScanner(bytes.NewBuffer(a)) bs := bufio.NewScanner(bytes.NewBuffer(b)) var failed bool for { if ab, bb := as.Scan(), bs.Scan(); ab != bb { t.Errorf("expect a & b to have same number of lines") return false } else if !ab { break } if v1, v2 := as.Text(), bs.Text(); v1 != v2 { t.Errorf("expect %q to be %q", v1, v2) failed = true } } return !failed }	2.99 KB	2025-05-25 12:27
📁 testdata	-	2025-05-25 12:27

Terminal