XTrim trims the stream to a certain size. It takes the name of the stream and the maximum length as arguments, and returns the number of messages in the stream after the trim operation. Parameters: name: The name of the stream. maxLen: The maximum length to trim the stream to. Returns: int: T
(name string, maxLen int)
| 502 | // - It encodes the streams and updates the stream in the database. |
| 503 | // - It returns the number of messages in the stream after the trim operation. |
| 504 | func (s *StreamStructure) XTrim(name string, maxLen int) (int, error) { |
| 505 | // Get the stream |
| 506 | encodedStreams, err := s.db.Get([]byte(name)) |
| 507 | if err != nil { |
| 508 | return 0, err |
| 509 | } |
| 510 | |
| 511 | // Decode the streams |
| 512 | if err = s.decodeStreams(encodedStreams, s.streams); err != nil { |
| 513 | return 0, err |
| 514 | } |
| 515 | |
| 516 | // Get the messages |
| 517 | messages := s.streams.Messages |
| 518 | |
| 519 | // Create a new slice of StreamMessage |
| 520 | var result []*StreamMessage |
| 521 | |
| 522 | // Get the messages |
| 523 | if len(messages) >= maxLen { |
| 524 | messages = messages[:maxLen] |
| 525 | // Convert []*StreamMessage to []StreamMessage |
| 526 | result = append(result, messages...) |
| 527 | } else { |
| 528 | return 0, ErrAmountOfData |
| 529 | } |
| 530 | |
| 531 | // Set the messages |
| 532 | s.streams.Messages = result |
| 533 | |
| 534 | // Encode the streams |
| 535 | encodedStreams, err = s.encodeStreams(s.streams) |
| 536 | if err != nil { |
| 537 | return 0, err |
| 538 | } |
| 539 | |
| 540 | // Set the stream |
| 541 | if err = s.db.Put([]byte(s.streams.Name), encodedStreams); err != nil { |
| 542 | return 0, err |
| 543 | } |
| 544 | |
| 545 | return len(s.streams.Messages), nil |
| 546 | } |
| 547 | |
| 548 | // XGroup creates a new consumer group. |
| 549 | // It takes the name of the stream, the name of the group, and the ID of the message as arguments, |