| | | 1 | | // Copyright (c) Microsoft Corporation. All rights reserved. |
| | | 2 | | // Licensed under the MIT License. |
| | | 3 | | |
| | | 4 | | using System.IO; |
| | | 5 | | using System.Threading; |
| | | 6 | | using System.Threading.Tasks; |
| | | 7 | | using Azure.Storage.Internal.Avro; |
| | | 8 | | |
| | | 9 | | namespace Azure.Storage.Blobs.ChangeFeed |
| | | 10 | | { |
| | | 11 | | internal class ChunkFactory |
| | | 12 | | { |
| | | 13 | | private readonly LazyLoadingBlobStreamFactory _lazyLoadingBlobStreamFactory; |
| | | 14 | | private readonly AvroReaderFactory _avroReaderFactory; |
| | | 15 | | private readonly BlobContainerClient _containerClient; |
| | | 16 | | |
| | 88 | 17 | | public ChunkFactory( |
| | 88 | 18 | | BlobContainerClient containerClient, |
| | 88 | 19 | | LazyLoadingBlobStreamFactory lazyLoadingBlobStreamFactory, |
| | 88 | 20 | | AvroReaderFactory avroReaderFactory) |
| | | 21 | | { |
| | 88 | 22 | | _containerClient = containerClient; |
| | 88 | 23 | | _lazyLoadingBlobStreamFactory = lazyLoadingBlobStreamFactory; |
| | 88 | 24 | | _avroReaderFactory = avroReaderFactory; |
| | 88 | 25 | | } |
| | | 26 | | |
| | | 27 | | internal async virtual Task<Chunk> BuildChunk( |
| | | 28 | | bool async, |
| | | 29 | | string chunkPath, |
| | | 30 | | long blockOffset = 0, |
| | | 31 | | long eventIndex = 0, |
| | | 32 | | CancellationToken cancellationToken = default) |
| | | 33 | | { |
| | 228 | 34 | | BlobClient blobClient = _containerClient.GetBlobClient(chunkPath); |
| | | 35 | | AvroReader avroReader; |
| | | 36 | | |
| | 228 | 37 | | Stream dataStream = _lazyLoadingBlobStreamFactory.BuildLazyLoadingBlobStream( |
| | 228 | 38 | | blobClient, |
| | 228 | 39 | | offset: blockOffset, |
| | 228 | 40 | | blockSize: Constants.ChangeFeed.ChunkBlockDownloadSize); |
| | | 41 | | |
| | | 42 | | // We aren't starting from the beginning of the Chunk |
| | 228 | 43 | | if (blockOffset != 0) |
| | | 44 | | { |
| | 52 | 45 | | Stream headStream = _lazyLoadingBlobStreamFactory.BuildLazyLoadingBlobStream( |
| | 52 | 46 | | blobClient, |
| | 52 | 47 | | offset: 0, |
| | 52 | 48 | | blockSize: Constants.ChangeFeed.LazyLoadingBlobStreamBlockSize); |
| | | 49 | | |
| | 52 | 50 | | avroReader = _avroReaderFactory.BuildAvroReader( |
| | 52 | 51 | | dataStream, |
| | 52 | 52 | | headStream, |
| | 52 | 53 | | blockOffset, |
| | 52 | 54 | | eventIndex); |
| | | 55 | | } |
| | | 56 | | else |
| | | 57 | | { |
| | 176 | 58 | | avroReader = _avroReaderFactory.BuildAvroReader(dataStream); |
| | | 59 | | } |
| | | 60 | | |
| | 228 | 61 | | await avroReader.Initalize(async, cancellationToken).ConfigureAwait(false); |
| | | 62 | | |
| | 228 | 63 | | return new Chunk( |
| | 228 | 64 | | avroReader, |
| | 228 | 65 | | blockOffset, |
| | 228 | 66 | | eventIndex, |
| | 228 | 67 | | chunkPath); |
| | 228 | 68 | | } |
| | | 69 | | |
| | | 70 | | /// <summary> |
| | | 71 | | /// Constructor for mocking. |
| | | 72 | | /// </summary> |
| | 40 | 73 | | public ChunkFactory() { } |
| | | 74 | | } |
| | | 75 | | } |