Add pieces announcement through DSN.

crates/subspace-service/src/dsn.rs

@@ -1,9 +1,10 @@
+use futures::future::join_all;
 use futures::StreamExt;
-use parity_scale_codec::Encode;
 use sc_consensus_subspace::{ArchivedSegmentNotification, SubspaceLink};
 use sp_core::traits::SpawnEssentialNamed;
 use sp_runtime::traits::Block as BlockT;
-use subspace_networking::{CreationError, PUB_SUB_ARCHIVING_TOPIC};
+use subspace_core_primitives::{PieceIndexHash, PIECES_IN_SEGMENT};
+use subspace_networking::CreationError;
 use tracing::{error, info, trace};
 
 /// Start an archiver that will listen for archived segments and send it to DSN network using
@@ -46,13 +47,24 @@ where
             }) = archived_segment_notification_stream.next().await
             {
                 trace!(target: "dsn", "ArchivedSegmentNotification received");
-                let data = archived_segment.encode().to_vec();
 
-                match node.publish(PUB_SUB_ARCHIVING_TOPIC.clone(), data).await {
-                    Ok(_) => {
+                let segment_index = archived_segment.root_block.segment_index();
+                let first_piece_index = segment_index * u64::from(PIECES_IN_SEGMENT);
+
+                let keys_iter = (first_piece_index..)
+                    .take(archived_segment.pieces.count())
+                    .map(PieceIndexHash::from_index)
+                    .map(|hash| hash.into());
+
+                let pieces_announcements = keys_iter
+                    .map(|key| node.announce_piece(key))
+                    .collect::<Vec<_>>();
+                let announcement_result = join_all(pieces_announcements).await;
+                match announcement_result.iter().find(|res| res.is_err()) {
+                    None => {
                         trace!(target: "dsn", "Archived segment published.");
                     }
-                    Err(err) => {
+                    Some(err) => {
                         error!(target: "dsn", error = ?err, "Failed to publish archived segment");
                     }
                 }