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
📄 error.go package eventstreamapi import ( "fmt" "sync" ) // InputWriterCloseErrorCode is used to denote an error occurred // while closing the event stream input writer. const InputWriterCloseErrorCode = "EventStreamInputWriterCloseError" type messageError struct { code string msg string } func (e messageError) Code() string { return e.code } func (e messageError) Message() string { return e.msg } func (e messageError) Error() string { return fmt.Sprintf("%s: %s", e.code, e.msg) } func (e messageError) OrigErr() error { return nil } // OnceError wraps the behavior of recording an error // once and signal on a channel when this has occurred. // Signaling is done by closing of the channel. // // Type is safe for concurrent usage. type OnceError struct { mu sync.RWMutex err error ch chan struct{} } // NewOnceError return a new OnceError func NewOnceError() OnceError { return &OnceError{ ch: make(chan struct{}, 1), } } // Err acquires a read-lock and returns an // error if one has been set. func (e OnceError) Err() error { e.mu.RLock() err := e.err e.mu.RUnlock() return err } // SetError acquires a write-lock and will set // the underlying error value if one has not been set. func (e OnceError) SetError(err error) { if err == nil { return } e.mu.Lock() if e.err == nil { e.err = err close(e.ch) } e.mu.Unlock() } // ErrorSet returns a channel that will be used to signal // that an error has been set. This channel will be closed // when the error value has been set for OnceError. func (e OnceError) ErrorSet() <-chan struct{} { return e.ch }	1.57 KB	2025-05-25 12:27
📄 reader.go package eventstreamapi import ( "fmt" "github.com/aws/aws-sdk-go/private/protocol" "github.com/aws/aws-sdk-go/private/protocol/eventstream" ) // Unmarshaler provides the interface for unmarshaling a EventStream // message into a SDK type. type Unmarshaler interface { UnmarshalEvent(protocol.PayloadUnmarshaler, eventstream.Message) error } // EventReader provides reading from the EventStream of an reader. type EventReader struct { decoder eventstream.Decoder unmarshalerForEventType func(string) (Unmarshaler, error) payloadUnmarshaler protocol.PayloadUnmarshaler payloadBuf []byte } // NewEventReader returns a EventReader built from the reader and unmarshaler // provided. Use ReadStream method to start reading from the EventStream. func NewEventReader( decoder eventstream.Decoder, payloadUnmarshaler protocol.PayloadUnmarshaler, unmarshalerForEventType func(string) (Unmarshaler, error), ) EventReader { return &EventReader{ decoder: decoder, payloadUnmarshaler: payloadUnmarshaler, unmarshalerForEventType: unmarshalerForEventType, payloadBuf: make([]byte, 101024), } } // ReadEvent attempts to read a message from the EventStream and return the // unmarshaled event value that the message is for. // // For EventStream API errors check if the returned error satisfies the // awserr.Error interface to get the error's Code and Message components. // // EventUnmarshalers called with EventStream messages must take copies of the // message's Payload. The payload will is reused between events read. func (r EventReader) ReadEvent() (event interface{}, err error) { msg, err := r.decoder.Decode(r.payloadBuf) if err != nil { return nil, err } defer func() { // Reclaim payload buffer for next message read. r.payloadBuf = msg.Payload[0:0] }() typ, err := GetHeaderString(msg, MessageTypeHeader) if err != nil { return nil, err } switch typ { case EventMessageType: return r.unmarshalEventMessage(msg) case ExceptionMessageType: return nil, r.unmarshalEventException(msg) case ErrorMessageType: return nil, r.unmarshalErrorMessage(msg) default: return nil, &UnknownMessageTypeError{ Type: typ, Message: msg.Clone(), } } } // UnknownMessageTypeError provides an error when a message is received from // the stream, but the reader is unable to determine what kind of message it is. type UnknownMessageTypeError struct { Type string Message eventstream.Message } func (e UnknownMessageTypeError) Error() string { return "unknown eventstream message type, " + e.Type } func (r EventReader) unmarshalEventMessage( msg eventstream.Message, ) (event interface{}, err error) { eventType, err := GetHeaderString(msg, EventTypeHeader) if err != nil { return nil, err } ev, err := r.unmarshalerForEventType(eventType) if err != nil { return nil, err } err = ev.UnmarshalEvent(r.payloadUnmarshaler, msg) if err != nil { return nil, err } return ev, nil } func (r EventReader) unmarshalEventException( msg eventstream.Message, ) (err error) { eventType, err := GetHeaderString(msg, ExceptionTypeHeader) if err != nil { return err } ev, err := r.unmarshalerForEventType(eventType) if err != nil { return err } err = ev.UnmarshalEvent(r.payloadUnmarshaler, msg) if err != nil { return err } var ok bool err, ok = ev.(error) if !ok { err = messageError{ code: "SerializationError", msg: fmt.Sprintf( "event stream exception %s mapped to non-error %T, %v", eventType, ev, ev, ), } } return err } func (r *EventReader) unmarshalErrorMessage(msg eventstream.Message) (err error) { var msgErr messageError msgErr.code, err = GetHeaderString(msg, ErrorCodeHeader) if err != nil { return err } msgErr.msg, err = GetHeaderString(msg, ErrorMessageHeader) if err != nil { return err } return msgErr } // GetHeaderString returns the value of the header as a string. If the header // is not set or the value is not a string an error will be returned. func GetHeaderString(msg eventstream.Message, headerName string) (string, error) { headerVal := msg.Headers.Get(headerName) if headerVal == nil { return "", fmt.Errorf("error header %s not present", headerName) } v, ok := headerVal.Get().(string) if !ok { return "", fmt.Errorf("error header value is not a string, %T", headerVal) } return v, nil }	4.3 KB	2025-05-25 12:27
📄 reader_test.go package eventstreamapi import ( "bytes" "fmt" "io" "testing" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/private/protocol" "github.com/aws/aws-sdk-go/private/protocol/eventstream" "github.com/aws/aws-sdk-go/private/protocol/restjson" ) var eventMessageTypeHeader = eventstream.Header{ Name: MessageTypeHeader, Value: eventstream.StringValue(EventMessageType), } func TestEventReader(t testing.T) { stream := createStream( eventstream.Message{ Headers: eventstream.Headers{ eventMessageTypeHeader, eventstream.Header{ Name: EventTypeHeader, Value: eventstream.StringValue("eventABC"), }, }, }, eventstream.Message{ Headers: eventstream.Headers{ eventMessageTypeHeader, eventstream.Header{ Name: EventTypeHeader, Value: eventstream.StringValue("eventEFG"), }, }, }, ) var unmarshalers request.HandlerList unmarshalers.PushBackNamed(restjson.UnmarshalHandler) decoder := eventstream.NewDecoder(stream) eventReader := NewEventReader(decoder, protocol.HandlerPayloadUnmarshal{ Unmarshalers: unmarshalers, }, unmarshalerForEventType, ) event, err := eventReader.ReadEvent() if err != nil { t.Fatalf("expect no error, got %v", err) } if event == nil { t.Fatalf("expect event got none") } event, err = eventReader.ReadEvent() if err == nil { t.Fatalf("expect error for unknown event, got none") } if event != nil { t.Fatalf("expect no event, got %T, %v", event, event) } } func TestEventReader_Error(t testing.T) { stream := createStream( eventstream.Message{ Headers: eventstream.Headers{ eventstream.Header{ Name: MessageTypeHeader, Value: eventstream.StringValue(ErrorMessageType), }, eventstream.Header{ Name: ErrorCodeHeader, Value: eventstream.StringValue("errorCode"), }, eventstream.Header{ Name: ErrorMessageHeader, Value: eventstream.StringValue("error message occur"), }, }, }, ) var unmarshalers request.HandlerList unmarshalers.PushBackNamed(restjson.UnmarshalHandler) decoder := eventstream.NewDecoder(stream) eventReader := NewEventReader(decoder, protocol.HandlerPayloadUnmarshal{ Unmarshalers: unmarshalers, }, unmarshalerForEventType, ) event, err := eventReader.ReadEvent() if err == nil { t.Fatalf("expect error got none") } if event != nil { t.Fatalf("expect no event, got %v", event) } if e, a := "errorCode: error message occur", err.Error(); e != a { t.Errorf("expect %v error, got %v", e, a) } } func TestEventReader_Exception(t testing.T) { eventMsgs := []eventstream.Message{ { Headers: eventstream.Headers{ eventstream.Header{ Name: MessageTypeHeader, Value: eventstream.StringValue(ExceptionMessageType), }, eventstream.Header{ Name: ExceptionTypeHeader, Value: eventstream.StringValue("exception"), }, }, Payload: []byte(`{"message":"exception message"}`), }, } stream := createStream(eventMsgs...) var unmarshalers request.HandlerList unmarshalers.PushBackNamed(restjson.UnmarshalHandler) decoder := eventstream.NewDecoder(stream) eventReader := NewEventReader(decoder, protocol.HandlerPayloadUnmarshal{ Unmarshalers: unmarshalers, }, unmarshalerForEventType, ) event, err := eventReader.ReadEvent() if err == nil { t.Fatalf("expect error got none") } if event != nil { t.Fatalf("expect no event, got %v", event) } et := err.(exceptionType) if e, a := string(eventMsgs[0].Payload), string(et.Payload); e != a { t.Errorf("expect %v payload, got %v", e, a) } } func BenchmarkEventReader(b testing.B) { var buf bytes.Buffer encoder := eventstream.NewEncoder(&buf) msg := eventstream.Message{ Headers: eventstream.Headers{ eventMessageTypeHeader, eventstream.Header{ Name: EventTypeHeader, Value: eventstream.StringValue("eventStructured"), }, }, Payload: []byte(`{"String":"stringfield","Number":123,"Nested":{"String":"fieldstring","Number":321}}`), } if err := encoder.Encode(msg); err != nil { b.Fatalf("failed to encode message, %v", err) } stream := bytes.NewReader(buf.Bytes()) var unmarshalers request.HandlerList unmarshalers.PushBackNamed(restjson.UnmarshalHandler) decoder := eventstream.NewDecoder(stream) eventReader := NewEventReader(decoder, protocol.HandlerPayloadUnmarshal{ Unmarshalers: unmarshalers, }, unmarshalerForEventType, ) b.ResetTimer() for i := 0; i < b.N; i++ { stream.Seek(0, 0) event, err := eventReader.ReadEvent() if err != nil { b.Fatalf("expect no error, got %v", err) } if event == nil { b.Fatalf("expect event got none") } } } func unmarshalerForEventType(eventType string) (Unmarshaler, error) { switch eventType { case "eventABC": return &eventABC{}, nil case "eventStructured": return &eventStructured{}, nil case "exception": return &exceptionType{}, nil default: return nil, fmt.Errorf("unknown event type, %v", eventType) } } type eventABC struct { _ struct{} HeaderField string Payload []byte } func (e eventABC) UnmarshalEvent( unmarshaler protocol.PayloadUnmarshaler, msg eventstream.Message, ) error { return nil } func createStream(msgs ...eventstream.Message) io.Reader { w := bytes.NewBuffer(nil) encoder := eventstream.NewEncoder(w) for _, msg := range msgs { if err := encoder.Encode(msg); err != nil { panic("createStream failed, " + err.Error()) } } return w } type exceptionType struct { Payload []byte } func (e exceptionType) Error() string { return fmt.Sprintf("exception error message") } func (e *exceptionType) UnmarshalEvent( unmarshaler protocol.PayloadUnmarshaler, msg eventstream.Message, ) error { e.Payload = msg.Payload return nil }	5.65 KB	2025-05-25 12:27
📄 shared.go package eventstreamapi // EventStream headers with specific meaning to async API functionality. const ( ChunkSignatureHeader = `:chunk-signature` // chunk signature for message DateHeader = `:date` // Date header for signature // Message header and values MessageTypeHeader = `:message-type` // Identifies type of message. EventMessageType = `event` ErrorMessageType = `error` ExceptionMessageType = `exception` // Message Events EventTypeHeader = `:event-type` // Identifies message event type e.g. "Stats". // Message Error ErrorCodeHeader = `:error-code` ErrorMessageHeader = `:error-message` // Message Exception ExceptionTypeHeader = `:exception-type` )	713 B	2025-05-25 12:27
📄 shared_test.go package eventstreamapi import ( "bytes" "encoding/hex" "time" "github.com/aws/aws-sdk-go/private/protocol" "github.com/aws/aws-sdk-go/private/protocol/eventstream" ) type mockChunkSigner struct { signature string err error } func (m mockChunkSigner) GetSignature(_, _ []byte, _ time.Time) ([]byte, error) { return mustDecodeBytes(hex.DecodeString(m.signature)), m.err } type eventStructured struct { _ struct{} `type:"structure"` String string `type:"string"` Number int64 `type:"long"` Nested eventStructured `type:"structure"` } func (e eventStructured) MarshalEvent(pm protocol.PayloadMarshaler) (eventstream.Message, error) { var msg eventstream.Message msg.Headers.Set(MessageTypeHeader, eventstream.StringValue(EventMessageType)) var buf bytes.Buffer if err := pm.MarshalPayload(&buf, e); err != nil { return eventstream.Message{}, err } msg.Payload = buf.Bytes() return msg, nil } func (e *eventStructured) UnmarshalEvent(pm protocol.PayloadUnmarshaler, msg eventstream.Message) error { return pm.UnmarshalPayload(bytes.NewReader(msg.Payload), e) } func mustDecodeBytes(b []byte, err error) []byte { if err != nil { panic(err) } return b } func swapTimeNow(f func() time.Time) func() { if f == nil { return func() {} } timeNow = f return func() { timeNow = time.Now } }	1.33 KB	2025-05-25 12:27
📄 signer.go package eventstreamapi import ( "bytes" "strings" "time" "github.com/aws/aws-sdk-go/private/protocol/eventstream" ) var timeNow = time.Now // StreamSigner defines an interface for the implementation of signing of event stream payloads type StreamSigner interface { GetSignature(headers, payload []byte, date time.Time) ([]byte, error) } // SignEncoder envelopes event stream messages // into an event stream message payload with included // signature headers using the provided signer and encoder. type SignEncoder struct { signer StreamSigner encoder Encoder bufEncoder BufferEncoder closeErr error closed bool } // NewSignEncoder returns a new SignEncoder using the provided stream signer and // event stream encoder. func NewSignEncoder(signer StreamSigner, encoder Encoder) SignEncoder { // TODO: Need to pass down logging return &SignEncoder{ signer: signer, encoder: encoder, bufEncoder: NewBufferEncoder(), } } // Close encodes a final event stream signing envelope with an empty event stream // payload. This final end-frame is used to mark the conclusion of the stream. func (s SignEncoder) Close() error { if s.closed { return s.closeErr } if err := s.encode([]byte{}); err != nil { if strings.Contains(err.Error(), "on closed pipe") { return nil } s.closeErr = err s.closed = true return s.closeErr } return nil } // Encode takes the provided message and add envelopes the message // with the required signature. func (s SignEncoder) Encode(msg eventstream.Message) error { payload, err := s.bufEncoder.Encode(msg) if err != nil { return err } return s.encode(payload) } func (s SignEncoder) encode(payload []byte) error { date := timeNow() var msg eventstream.Message msg.Headers.Set(DateHeader, eventstream.TimestampValue(date)) msg.Payload = payload var headers bytes.Buffer if err := eventstream.EncodeHeaders(&headers, msg.Headers); err != nil { return err } sig, err := s.signer.GetSignature(headers.Bytes(), msg.Payload, date) if err != nil { return err } msg.Headers.Set(ChunkSignatureHeader, eventstream.BytesValue(sig)) return s.encoder.Encode(msg) } // BufferEncoder is a utility that provides a buffered // event stream encoder type BufferEncoder struct { encoder Encoder buffer bytes.Buffer } // NewBufferEncoder returns a new BufferEncoder initialized // with a 1024 byte buffer. func NewBufferEncoder() BufferEncoder { buf := bytes.NewBuffer(make([]byte, 1024)) return &BufferEncoder{ encoder: eventstream.NewEncoder(buf), buffer: buf, } } // Encode returns the encoded message as a byte slice. // The returned byte slice will be modified on the next encode call // and should not be held onto. func (e *BufferEncoder) Encode(msg eventstream.Message) ([]byte, error) { e.buffer.Reset() if err := e.encoder.Encode(msg); err != nil { return nil, err } return e.buffer.Bytes(), nil }	2.86 KB	2025-05-25 12:27
📄 signer_test.go //go:build go1.7 // +build go1.7 package eventstreamapi import ( "bytes" "encoding/base64" "encoding/hex" "fmt" "strings" "testing" "time" "github.com/aws/aws-sdk-go/private/protocol/eventstream" "github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamtest" ) func TestSignEncoder(t testing.T) { currentTime := time.Date(2019, 1, 27, 22, 37, 54, 0, time.UTC) cases := map[string]struct { Signer StreamSigner Input eventstream.Message Expect eventstream.Message NestedExpect eventstream.Message Err string }{ "sign message": { Signer: mockChunkSigner{ signature: "524f1d03d1d81e94a099042736d40bd9681b867321443ff58a4568e274dbd83bff", }, Input: eventstream.Message{ Headers: eventstream.Headers{ { Name: "header_name", Value: eventstream.StringValue("header value"), }, }, Payload: []byte("payload"), }, Expect: eventstream.Message{ Headers: eventstream.Headers{ { Name: ":date", Value: eventstream.TimestampValue(currentTime), }, { Name: ":chunk-signature", Value: eventstream.BytesValue(mustDecodeBytes( hex.DecodeString("524f1d03d1d81e94a099042736d40bd9681b867321443ff58a4568e274dbd83bff"), )), }, }, Payload: mustDecodeBytes( base64.StdEncoding.DecodeString( `AAAAMgAAABs0pv1jC2hlYWRlcl9uYW1lBwAMaGVhZGVyIHZhbHVlcGF5bG9hZH4tKFg=`, ), ), }, NestedExpect: &eventstream.Message{ Headers: eventstream.Headers{ { Name: "header_name", Value: eventstream.StringValue("header value"), }, }, Payload: []byte(`payload`), }, }, "signing error": { Signer: mockChunkSigner{err: fmt.Errorf("signing error")}, Input: eventstream.Message{ Headers: []eventstream.Header{ { Name: "header_name", Value: eventstream.StringValue("header value"), }, }, Payload: []byte("payload"), }, Err: "signing error", }, } origNowFn := timeNow timeNow = func() time.Time { return currentTime } defer func() { timeNow = origNowFn }() decodeBuf := make([]byte, 1024) for name, c := range cases { t.Run(name, func(t testing.T) { encoder := &mockEncoder{} signer := NewSignEncoder(c.Signer, encoder) err := signer.Encode(c.Input) if err == nil && len(c.Err) > 0 { t.Fatalf("expected error, but got nil") } else if err != nil && len(c.Err) == 0 { t.Fatalf("expected no error, but got %v", err) } else if err != nil && len(c.Err) > 0 && !strings.Contains(err.Error(), c.Err) { t.Fatalf("expected %v, but got %v", c.Err, err) } else if len(c.Err) > 0 { return } eventstreamtest.AssertMessageEqual(t, c.Expect, encoder.msgs[0], "envelope msg") if c.NestedExpect != nil { nested := eventstream.NewDecoder(bytes.NewReader(encoder.msgs[0].Payload)) nestedMsg, err := nested.Decode(decodeBuf) if err != nil { t.Fatalf("expect no decode error got, %v", err) } eventstreamtest.AssertMessageEqual(t, c.NestedExpect, nestedMsg, "nested msg") } }) } } type mockEncoder struct { msgs []eventstream.Message } func (m *mockEncoder) Encode(msg eventstream.Message) error { m.msgs = append(m.msgs, msg) return nil }	3.19 KB	2025-05-25 12:27
📄 stream_writer.go package eventstreamapi import ( "fmt" "io" "sync" "github.com/aws/aws-sdk-go/aws" ) // StreamWriter provides concurrent safe writing to an event stream. type StreamWriter struct { eventWriter EventWriter stream chan eventWriteAsyncReport done chan struct{} closeOnce sync.Once err OnceError streamCloser io.Closer } // NewStreamWriter returns a StreamWriter for the event writer, and stream // closer provided. func NewStreamWriter(eventWriter EventWriter, streamCloser io.Closer) StreamWriter { w := &StreamWriter{ eventWriter: eventWriter, streamCloser: streamCloser, stream: make(chan eventWriteAsyncReport), done: make(chan struct{}), err: NewOnceError(), } go w.writeStream() return w } // Close terminates the writers ability to write new events to the stream. Any // future call to Send will fail with an error. func (w StreamWriter) Close() error { w.closeOnce.Do(w.safeClose) return w.Err() } func (w StreamWriter) safeClose() { close(w.done) } // ErrorSet returns a channel which will be closed // if an error occurs. func (w StreamWriter) ErrorSet() <-chan struct{} { return w.err.ErrorSet() } // Err returns any error that occurred while attempting to write an event to the // stream. func (w StreamWriter) Err() error { return w.err.Err() } // Send writes a single event to the stream returning an error if the write // failed. // // Send may be called concurrently. Events will be written to the stream // safely. func (w StreamWriter) Send(ctx aws.Context, event Marshaler) error { if err := w.Err(); err != nil { return err } resultCh := make(chan error) wrapped := eventWriteAsyncReport{ Event: event, Result: resultCh, } select { case w.stream <- wrapped: case <-ctx.Done(): return ctx.Err() case <-w.done: return fmt.Errorf("stream closed, unable to send event") } select { case err := <-resultCh: return err case <-ctx.Done(): return ctx.Err() case <-w.done: return fmt.Errorf("stream closed, unable to send event") } } func (w StreamWriter) writeStream() { defer w.Close() for { select { case wrapper := <-w.stream: err := w.eventWriter.WriteEvent(wrapper.Event) wrapper.ReportResult(w.done, err) if err != nil { w.err.SetError(err) return } case <-w.done: if err := w.streamCloser.Close(); err != nil { w.err.SetError(err) } return } } } type eventWriteAsyncReport struct { Event Marshaler Result chan<- error } func (e eventWriteAsyncReport) ReportResult(cancel <-chan struct{}, err error) bool { select { case e.Result <- err: return true case <-cancel: return false } }	2.62 KB	2025-05-25 12:27
📄 transport.go //go:build go1.18 // +build go1.18 package eventstreamapi import "github.com/aws/aws-sdk-go/aws/request" // ApplyHTTPTransportFixes is a no-op for Go 1.18 and above. func ApplyHTTPTransportFixes(r *request.Request) { }	222 B	2025-05-25 12:27
📄 transport_go1.17.go //go:build !go1.18 // +build !go1.18 package eventstreamapi import "github.com/aws/aws-sdk-go/aws/request" // ApplyHTTPTransportFixes applies fixes to the HTTP request for proper event // stream functionality. Go 1.15 through 1.17 HTTP client could hang forever // when an HTTP/2 connection failed with an non-200 status code and err. Using // Expect 100-Continue, allows the HTTP client to gracefully handle the non-200 // status code, and close the connection. // // This is a no-op for Go 1.18 and above. func ApplyHTTPTransportFixes(r request.Request) { r.Handlers.Sign.PushBack(func(r request.Request) { r.HTTPRequest.Header.Set("Expect", "100-Continue") }) }	674 B	2025-05-25 12:27
📄 writer.go package eventstreamapi import ( "github.com/aws/aws-sdk-go/private/protocol" "github.com/aws/aws-sdk-go/private/protocol/eventstream" ) // Marshaler provides a marshaling interface for event types to event stream // messages. type Marshaler interface { MarshalEvent(protocol.PayloadMarshaler) (eventstream.Message, error) } // Encoder is an stream encoder that will encode an event stream message for // the transport. type Encoder interface { Encode(eventstream.Message) error } // EventWriter provides a wrapper around the underlying event stream encoder // for an io.WriteCloser. type EventWriter struct { encoder Encoder payloadMarshaler protocol.PayloadMarshaler eventTypeFor func(Marshaler) (string, error) } // NewEventWriter returns a new event stream writer, that will write to the // writer provided. Use the WriteEvent method to write an event to the stream. func NewEventWriter(encoder Encoder, pm protocol.PayloadMarshaler, eventTypeFor func(Marshaler) (string, error), ) EventWriter { return &EventWriter{ encoder: encoder, payloadMarshaler: pm, eventTypeFor: eventTypeFor, } } // WriteEvent writes an event to the stream. Returns an error if the event // fails to marshal into a message, or writing to the underlying writer fails. func (w EventWriter) WriteEvent(event Marshaler) error { msg, err := w.marshal(event) if err != nil { return err } return w.encoder.Encode(msg) } func (w *EventWriter) marshal(event Marshaler) (eventstream.Message, error) { eventType, err := w.eventTypeFor(event) if err != nil { return eventstream.Message{}, err } msg, err := event.MarshalEvent(w.payloadMarshaler) if err != nil { return eventstream.Message{}, err } msg.Headers.Set(EventTypeHeader, eventstream.StringValue(eventType)) return msg, nil }	1.78 KB	2025-05-25 12:27
📄 writer_test.go //go:build go1.7 // +build go1.7 package eventstreamapi import ( "bytes" "encoding/base64" "encoding/hex" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/private/protocol" "github.com/aws/aws-sdk-go/private/protocol/eventstream" "github.com/aws/aws-sdk-go/private/protocol/eventstream/eventstreamtest" "github.com/aws/aws-sdk-go/private/protocol/restjson" ) func TestEventWriter(t testing.T) { cases := map[string]struct { Event Marshaler EncodeWrapper func(e Encoder) Encoder TimeFunc func() time.Time Expect eventstream.Message NestedExpect eventstream.Message }{ "structured event": { Event: &eventStructured{ String: aws.String("stringfield"), Number: aws.Int64(123), Nested: &eventStructured{ String: aws.String("fieldstring"), Number: aws.Int64(321), }, }, Expect: eventstream.Message{ Headers: eventstream.Headers{ eventMessageTypeHeader, eventstream.Header{ Name: EventTypeHeader, Value: eventstream.StringValue("eventStructured"), }, }, Payload: []byte(`{"String":"stringfield","Number":123,"Nested":{"String":"fieldstring","Number":321}}`), }, }, "signed event": { Event: &eventStructured{ String: aws.String("stringfield"), Number: aws.Int64(123), Nested: &eventStructured{ String: aws.String("fieldstring"), Number: aws.Int64(321), }, }, EncodeWrapper: func(e Encoder) Encoder { return NewSignEncoder( &mockChunkSigner{ signature: "524f1d03d1d81e94a099042736d40bd9681b867321443ff58a4568e274dbd83bff", }, e, ) }, TimeFunc: func() time.Time { return time.Date(2019, 1, 27, 22, 37, 54, 0, time.UTC) }, Expect: eventstream.Message{ Headers: eventstream.Headers{ { Name: DateHeader, Value: eventstream.TimestampValue(time.Date(2019, 1, 27, 22, 37, 54, 0, time.UTC)), }, { Name: ChunkSignatureHeader, Value: eventstream.BytesValue(mustDecodeBytes( hex.DecodeString("524f1d03d1d81e94a099042736d40bd9681b867321443ff58a4568e274dbd83bff"), )), }, }, Payload: mustDecodeBytes(base64.StdEncoding.DecodeString( `AAAAmAAAADSl4EcNDTptZXNzYWdlLXR5cGUHAAVldmVudAs6ZXZlbnQtdHlwZQcAD2V2ZW50U3RydWN0dXJlZHsiU3RyaW5nIjoic3RyaW5nZmllbGQiLCJOdW1iZXIiOjEyMywiTmVzdGVkIjp7IlN0cmluZyI6ImZpZWxkc3RyaW5nIiwiTnVtYmVyIjozMjF9fdVW3Ow=`, )), }, }, } var marshalers request.HandlerList marshalers.PushBackNamed(restjson.BuildHandler) var stream bytes.Buffer decodeBuf := make([]byte, 1024) for name, c := range cases { t.Run(name, func(t testing.T) { defer swapTimeNow(c.TimeFunc)() stream.Reset() var encoder Encoder encoder = eventstream.NewEncoder(&stream, eventstream.EncodeWithLogger(t)) if c.EncodeWrapper != nil { encoder = c.EncodeWrapper(encoder) } eventWriter := NewEventWriter(encoder, protocol.HandlerPayloadMarshal{ Marshalers: marshalers, }, func(event Marshaler) (string, error) { return "eventStructured", nil }, ) decoder := eventstream.NewDecoder(&stream) if err := eventWriter.WriteEvent(c.Event); err != nil { t.Fatalf("expect no write error, got %v", err) } msg, err := decoder.Decode(decodeBuf) if err != nil { t.Fatalf("expect no decode error got, %v", err) } eventstreamtest.AssertMessageEqual(t, c.Expect, msg) }) } } func BenchmarkEventWriter(b testing.B) { var marshalers request.HandlerList marshalers.PushBackNamed(restjson.BuildHandler) var stream bytes.Buffer encoder := eventstream.NewEncoder(&stream) eventWriter := NewEventWriter(encoder, protocol.HandlerPayloadMarshal{ Marshalers: marshalers, }, func(event Marshaler) (string, error) { return "eventStructured", nil }, ) event := &eventStructured{ String: aws.String("stringfield"), Number: aws.Int64(123), Nested: &eventStructured{ String: aws.String("fieldstring"), Number: aws.Int64(321), }, } b.ResetTimer() for i := 0; i < b.N; i++ { if err := eventWriter.WriteEvent(event); err != nil { b.Fatalf("expect no write error, got %v", err) } } }	4.13 KB	2025-05-25 12:27

Terminal