Method XAdd

structure/stream.go:98–171

XAdd adds a new message to a stream. If the stream does not exist, it will be created. If the message ID already exists, it will return false. Parameters: name: The name of the stream. id: The ID of the message. fields: The fields (attributes) of the message. It should be a map[string]

(name, id string, fields map[string]interface{})

// - If the message ID already exists in the stream, the function will return false and ErrExistID.
// - Otherwise, the message will be added to the stream, and the stream will be stored in the database.
func (s *StreamStructure) XAdd(name, id string, fields map[string]interface{}) (bool, error) {
	// Check if the arguments are valid
	if len(id) == 0 || len(fields) == 0 || fields == nil {
		return false, ErrInvalidXArgs
	}

	// Get the stream
	value, _ := s.db.Get([]byte(name))
	if value == nil {
		// Create a new stream
		s.streams = &Streams{
			Name:        name,
			Messages:    []*StreamMessage{},
			Groups:      map[string]*StreamGroup{},
			LastMessage: time.Now(),
		}

		// Create a new message
		message := &StreamMessage{
			Id:     id,
			Fields: fields,
		}
		// Add the message to the stream
		s.streams.Messages = append(s.streams.Messages, message)

		// Encode the streams
		encodedStreams, err := s.encodeStreams(s.streams)
		if err != nil {
			return false, err
		}

		// Set the stream
		if err = s.db.Put([]byte(s.streams.Name), encodedStreams); err != nil {
			return false, err
		}

		return true, nil
	} else {
		// Decode the streams
		err := s.decodeStreams(value, s.streams)
		if err != nil {
			return false, err
		}

		// Check if the message ID already exists
		for _, msg := range s.streams.Messages {
			if msg.Id == id {
				return false, ErrExistID
			}
		}

		// Create a new message
		message := &StreamMessage{
			Id:     id,
			Fields: fields,
		}

		// Add the message to the stream