chore: move file iterators to fluvio_storage crate (#3613)

crates/fluvio-spu/Cargo.toml

@@ -51,7 +51,7 @@ chrono = { workspace = true }
 # Fluvio dependencies
 fluvio = { workspace = true }
 fluvio-types = { workspace = true, features = ["events"] }
-fluvio-storage = { workspace = true }
+fluvio-storage = { workspace = true, features = ["iterators"] }
 fluvio-compression = { workspace = true }
 fluvio-controlplane = { workspace = true }
 fluvio-controlplane-metadata = { workspace = true }

crates/fluvio-spu/src/services/public/stream_fetch.rs

@@ -4,16 +4,18 @@ use std::time::Instant;
 use tracing::{debug, error, instrument, trace, warn};
 use tokio::select;
 
+use fluvio_compression::CompressionError;
 use fluvio_controlplane_metadata::partition::ReplicaKey;
 use fluvio_types::event::{StickyEvent, offsets::OffsetPublisher};
 use fluvio_future::task::spawn;
-use fluvio_socket::{ExclusiveFlvSink, SocketError};
 use fluvio_protocol::{
     api::{RequestMessage, RequestHeader},
     record::{RecordSet, Offset, RawRecords},
 };
 use fluvio_protocol::link::{ErrorCode, smartmodule::SmartModuleTransformRuntimeError};
-use fluvio_compression::CompressionError;
+use fluvio_protocol::record::Batch;
+use fluvio_socket::{ExclusiveFlvSink, SocketError};
+use fluvio_storage::iterators::FileBatchIterator;
 use fluvio_spu_schema::{
     server::stream_fetch::{
         DefaultStreamFetchRequest, FileStreamFetchRequest, StreamFetchRequest, StreamFetchResponse,
@@ -23,15 +25,13 @@ use fluvio_spu_schema::{
     file::FileRecordSet,
 };
 use fluvio_types::event::offsets::OffsetChangeListener;
-use fluvio_protocol::record::Batch;
 
 use crate::core::{DefaultSharedGlobalContext, metrics::IncreaseValue};
 use crate::replication::leader::SharedFileLeaderState;
 use crate::services::public::conn_context::ConnectionContext;
 use crate::services::public::stream_fetch::publishers::INIT_OFFSET;
 use crate::smartengine::context::SmartModuleContext;
 use crate::smartengine::batch::process_batch;
-use crate::smartengine::file_batch::FileBatchIterator;
 use crate::core::metrics::SpuMetrics;
 use crate::traffic::TrafficType;
 

crates/fluvio-spu/src/services/public/tests/produce.rs

@@ -3,7 +3,7 @@ use std::{env::temp_dir, time::Duration};
 use fluvio::{SmartModuleInvocation, SmartModuleInvocationWasm, SmartModuleKind};
 use fluvio_controlplane::replica::Replica;
 use fluvio_smartmodule::{Record, dataplane::smartmodule::Lookback};
-use fluvio_storage::FileReplica;
+use fluvio_storage::{FileReplica, iterators::FileBatchIterator};
 use tracing::debug;
 
 use fluvio_protocol::{
@@ -34,7 +34,6 @@ use crate::{
         },
     },
     replication::leader::LeaderReplicaState,
-    smartengine::file_batch::FileBatchIterator,
 };
 
 #[fluvio_future::test(ignore)]

crates/fluvio-spu/src/smartengine/context.rs

@@ -7,6 +7,7 @@ use fluvio_protocol::link::ErrorCode;
 use fluvio_smartmodule::Record;
 use fluvio_spu_schema::server::smartmodule::{SmartModuleInvocation, SmartModuleInvocationWasm};
 use fluvio_storage::ReplicaStorage;
+use fluvio_storage::iterators::{FileBatch, FileBatchIterator, FileRecordIterator, RecordItem};
 use fluvio_types::Timestamp;
 use tracing::{debug, trace, error};
 
@@ -15,14 +16,11 @@ use crate::core::metrics::SpuMetrics;
 use crate::replication::leader::LeaderReplicaState;
 
 use crate::smartengine::chain;
-use crate::smartengine::file_batch::{FileRecordIterator, FileBatchIterator};
 use crate::smartengine::Lookback;
 use crate::smartengine::SmartModuleChainBuilder;
 use crate::smartengine::SmartModuleChainInstance;
 use crate::smartengine::Version;
 
-use super::file_batch::{FileBatch, RecordItem};
-
 #[derive(Debug)]
 pub struct SmartModuleContext {
     chain: SmartModuleChainInstance,