| 622 | } |
| 623 | |
| 624 | func convertDirectGetMsgResponseToMsg(r *nats.Msg) (*RawStreamMsg, error) { |
| 625 | // Check for 404/408. We would get a no-payload message and a "Status" header |
| 626 | if len(r.Data) == 0 { |
| 627 | val := r.Header.Get(statusHdr) |
| 628 | if val != "" { |
| 629 | switch val { |
| 630 | case statusNoMsgs: |
| 631 | return nil, ErrMsgNotFound |
| 632 | default: |
| 633 | desc := r.Header.Get("Description") |
| 634 | if desc == "" { |
| 635 | desc = "unable to get message" |
| 636 | } |
| 637 | return nil, fmt.Errorf("nats: %s", desc) |
| 638 | } |
| 639 | } |
| 640 | } |
| 641 | |
| 642 | // Check for headers that give us the required information to |
| 643 | // reconstruct the message. |
| 644 | if len(r.Header) == 0 { |
| 645 | return nil, errors.New("nats: response should have headers") |
| 646 | } |
| 647 | stream := r.Header.Get(StreamHeader) |
| 648 | if stream == "" { |
| 649 | return nil, errors.New("nats: missing stream header") |
| 650 | } |
| 651 | |
| 652 | seqStr := r.Header.Get(SequenceHeader) |
| 653 | if seqStr == "" { |
| 654 | return nil, errors.New("nats: missing sequence header") |
| 655 | } |
| 656 | seq, err := strconv.ParseUint(seqStr, 10, 64) |
| 657 | if err != nil { |
| 658 | return nil, fmt.Errorf("nats: invalid sequence header '%s': %v", seqStr, err) |
| 659 | } |
| 660 | timeStr := r.Header.Get(TimeStampHeaer) |
| 661 | if timeStr == "" { |
| 662 | return nil, errors.New("nats: missing timestamp header") |
| 663 | } |
| 664 | |
| 665 | tm, err := time.Parse(time.RFC3339Nano, timeStr) |
| 666 | if err != nil { |
| 667 | return nil, fmt.Errorf("nats: invalid timestamp header '%s': %v", timeStr, err) |
| 668 | } |
| 669 | subj := r.Header.Get(SubjectHeader) |
| 670 | if subj == "" { |
| 671 | return nil, errors.New("nats: missing subject header") |
| 672 | } |
| 673 | return &RawStreamMsg{ |
| 674 | Subject: subj, |
| 675 | Sequence: seq, |
| 676 | Header: r.Header, |
| 677 | Data: r.Data, |
| 678 | Time: tm, |
| 679 | }, nil |
| 680 | } |
| 681 | |