readStreamEntryContent read messages in a stream entry
(buf []byte, cursor *int, firstId *model.StreamId)
| 124 | |
| 125 | // readStreamEntryContent read messages in a stream entry |
| 126 | func (dec *Decoder) readStreamEntryContent(buf []byte, cursor *int, firstId *model.StreamId) (*model.StreamEntry, error) { |
| 127 | // read count |
| 128 | count, err := dec.readListPackEntryAsInt(buf, cursor) |
| 129 | if err != nil { |
| 130 | return nil, fmt.Errorf("read stream entry count failed: %v", err) |
| 131 | } |
| 132 | deleted, err := dec.readListPackEntryAsInt(buf, cursor) |
| 133 | if err != nil { |
| 134 | return nil, fmt.Errorf("read stream entry deleted count failed: %v", err) |
| 135 | } |
| 136 | |
| 137 | // read field names of master entry |
| 138 | fieldNum0, err := dec.readListPackEntryAsInt(buf, cursor) |
| 139 | if err != nil { |
| 140 | return nil, fmt.Errorf("read stream field number failed: %v", err) |
| 141 | } |
| 142 | masterFieldNum := int(fieldNum0) |
| 143 | masterFieldNames := make([]string, masterFieldNum) |
| 144 | for i := 0; i < masterFieldNum; i++ { |
| 145 | name, err := dec.readListPackEntryAsString(buf, cursor) |
| 146 | if err != nil { |
| 147 | return nil, fmt.Errorf("read field name of stream entry failed: %v", err) |
| 148 | } |
| 149 | masterFieldNames[i] = string(name) |
| 150 | } |
| 151 | // read lp count of master entry |
| 152 | if _, err = dec.readListPackEntryAsString(buf, cursor); err != nil { |
| 153 | return nil, fmt.Errorf("read fields end flag failed: %v", err) |
| 154 | } |
| 155 | |
| 156 | total := count + deleted |
| 157 | msgs := make([]*model.StreamMessage, 0, total) |
| 158 | for i := int64(0); i < total; i++ { |
| 159 | flag, err := dec.readListPackEntryAsInt(buf, cursor) |
| 160 | if err != nil { |
| 161 | return nil, fmt.Errorf("read stream item flag failed: %v", err) |
| 162 | } |
| 163 | ms, err := dec.readListPackEntryAsInt(buf, cursor) |
| 164 | if err != nil { |
| 165 | return nil, fmt.Errorf("read stream item id ms failed: %v", err) |
| 166 | } |
| 167 | seq, err := dec.readListPackEntryAsInt(buf, cursor) |
| 168 | if err != nil { |
| 169 | return nil, fmt.Errorf("read stream item id seq failed: %v", err) |
| 170 | } |
| 171 | // ms and seq may be negative |
| 172 | msgId := &model.StreamId{ |
| 173 | Ms: uint64(ms + int64(firstId.Ms)), |
| 174 | Sequence: uint64(seq + int64(firstId.Sequence)), |
| 175 | } |
| 176 | fieldNum := masterFieldNum |
| 177 | if flag&StreamItemFlagSameFields == 0 { |
| 178 | fieldNum0, err := dec.readListPackEntryAsInt(buf, cursor) |
| 179 | if err != nil { |
| 180 | return nil, fmt.Errorf("read stream item field number failed: %v", err) |
| 181 | } |
| 182 | fieldNum = int(fieldNum0) |
| 183 | } |
no test coverage detected