(reader io.Reader, contentMetadata ContentMetadata, key *Key)
| 249 | } |
| 250 | |
| 251 | func (r *redisCache) Put(reader io.Reader, contentMetadata ContentMetadata, key *Key) (time.Duration, error) { |
| 252 | medatadata := r.encodeMetadata(&contentMetadata) |
| 253 | |
| 254 | stringKey := key.String() |
| 255 | // in order to make the streaming operation atomic, chproxy streams into a temporary key (only known by the current goroutine) |
| 256 | // then it switches the full result to the "real" stringKey available for other goroutines |
| 257 | // nolint:gosec // not security sensitve, only used internally. |
| 258 | random := strconv.Itoa(rand.Int()) |
| 259 | // Redis RENAME is considered to be a multikey operation. In Cluster mode, both oldkey and renamedkey must be in the same hash slot, |
| 260 | // Refer Redis Documentation here: https://redis.io/commands/rename/ |
| 261 | // To solve this,we need to force the temporary key to be in the same hash slot. We can do this by adding hashtag to the |
| 262 | // actual part of the temporary key. When the key contains a "{...}" pattern, only the substring between the braces, "{" and "}," |
| 263 | // is hashed to obtain the hash slot. |
| 264 | // Refer the hash tags section of Redis documentation here: https://redis.io/docs/reference/cluster-spec/#hash-tags |
| 265 | stringKeyTmp := "{" + stringKey + "}" + random + "_tmp" |
| 266 | |
| 267 | ctxSet, cancelFuncSet := context.WithTimeout(context.Background(), putTimeout) |
| 268 | defer cancelFuncSet() |
| 269 | err := r.client.Set(ctxSet, stringKeyTmp, medatadata, r.expire).Err() |
| 270 | if err != nil { |
| 271 | return 0, err |
| 272 | } |
| 273 | // we don't fetch all the reader content bulks by bulks to from redis to avoid memory issue |
| 274 | // if the content is big (which is the case when chproxy users are fetching a lot of data) |
| 275 | buffer := make([]byte, 2*1024*1024) |
| 276 | totalByteWrittenExpected := len(medatadata) |
| 277 | for { |
| 278 | n, err := reader.Read(buffer) |
| 279 | // the reader should return an err = io.EOF once it has nothing to read or at the last read call with content. |
| 280 | // But this is not the case with this reader so we check the condition n == 0 to exit the read loop. |
| 281 | // We kept the err == io.EOF in the loop in case the behavior of the reader changes |
| 282 | |
| 283 | if n == 0 { |
| 284 | break |
| 285 | } |
| 286 | if err != nil && !errors.Is(err, io.EOF) { |
| 287 | return 0, err |
| 288 | } |
| 289 | ctxAppend, cancelFuncAppend := context.WithTimeout(context.Background(), putTimeout) |
| 290 | defer cancelFuncAppend() |
| 291 | totalByteWritten, err := r.client.Append(ctxAppend, stringKeyTmp, string(buffer[:n])).Result() |
| 292 | if err != nil { |
| 293 | // trying to clean redis from this partially inserted item |
| 294 | r.clean(stringKeyTmp) |
| 295 | return 0, err |
| 296 | } |
| 297 | totalByteWrittenExpected += n |
| 298 | if int(totalByteWritten) != totalByteWrittenExpected { |
| 299 | // trying to clean redis from this partially inserted item |
| 300 | r.clean(stringKeyTmp) |
| 301 | return 0, fmt.Errorf("could not stream the value into redis, only %d bytes were written instead of %d", totalByteWritten, totalByteWrittenExpected) |
| 302 | } |
| 303 | if errors.Is(err, io.EOF) { |
| 304 | break |
| 305 | } |
| 306 | } |
| 307 | // at this step we know that the item stored in stringKeyTmp is fully written |
| 308 | // so we can put it to its final stringKey |
nothing calls this directly
no test coverage detected