XGroup creates a new consumer group. It takes the name of the stream, the name of the group, and the ID of the message as arguments, and returns a boolean indicating whether the group was created successfully or not, and an error if any. Parameters: name: The name of the stream. group: The name
(name, group, id string)
| 573 | // - It returns true if the group was created successfully, and false otherwise. |
| 574 | // - If any error occurs during the operation, it will be returned along with false. |
| 575 | func (s *StreamStructure) XGroup(name, group, id string) (bool, error) { |
| 576 | // Get the stream |
| 577 | encodedStreams, err := s.db.Get([]byte(name)) |
| 578 | if err != nil { |
| 579 | return false, err |
| 580 | } |
| 581 | |
| 582 | // Decode the streams |
| 583 | if err = s.decodeStreams(encodedStreams, s.streams); err != nil { |
| 584 | return false, err |
| 585 | } |
| 586 | |
| 587 | // Create a new stream group |
| 588 | sg := &StreamGroup{ |
| 589 | Name: group, |
| 590 | LastGeneratedID: id, |
| 591 | LastDeliveredTime: time.Now(), |
| 592 | PendingMessages: make(map[string]*StreamMessage), |
| 593 | } |
| 594 | |
| 595 | // Set the stream group |
| 596 | s.streams.Groups[group] = sg |
| 597 | |
| 598 | // Encode the streams |
| 599 | encodedStreams, err = s.encodeStreams(s.streams) |
| 600 | if err != nil { |
| 601 | return false, err |
| 602 | } |
| 603 | |
| 604 | // Set the stream |
| 605 | if err = s.db.Put([]byte(s.streams.Name), encodedStreams); err != nil { |
| 606 | return false, err |
| 607 | } |
| 608 | |
| 609 | return true, nil |
| 610 | } |
| 611 | |
| 612 | // encodeStreams encodes the streams. |
| 613 | // It takes the Streams object as an argument and returns the encoded data as a byte slice. |