You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: p2p_service.rs
    path of file: ./repos/fuel-core/crates/services/p2p/src
    
    the code of the file until where you have to start completion: use crate::{
    behavior::{
        FuelBehaviour,
        FuelBehaviourEvent,
    },
    codecs::{
        postcard::PostcardCodec,
        GossipsubCodec,
    },
    config::{
        build_transport_function,
        Config,
    },
    gossipsub::{
        messages::{
            GossipsubBroadcastRequest,
            GossipsubMessage as FuelGossipsubMessage,
        },
        topics::GossipsubTopics,
    },
    heartbeat,
    peer_manager::{
        PeerManager,
        Punisher,
    },
    peer_report::PeerReportEvent,
    request_response::messages::{
        RequestError,
        RequestMessage,
        ResponseError,
        ResponseMessage,
        ResponseSendError,
        ResponseSender,
    },
    TryPeerId,
};
use fuel_core_metrics::p2p_metrics::p2p_metrics;
use fuel_core_types::{
    fuel_types::BlockHeight,
    services::p2p::peer_reputation::AppScore,
};
use futures::prelude::*;
use libp2p::{
    gossipsub::{
        self,
        MessageAcceptance,
        MessageId,
        PublishError,
        TopicHash,
    },
    identify,
    multiaddr::Protocol,
    request_response::{
        self,
        InboundRequestId,
        OutboundRequestId,
        ResponseChannel,
    },
    swarm::SwarmEvent,
    Multiaddr,
    PeerId,
    Swarm,
    SwarmBu
    
    async fn request_response_works_with(request_msg: RequestMessage) {
        let mut p2p_config = Config::default_initialized("request_response_works_with");

        // Node A
        let mut node_a = build_service_from_config(p2p_config.clone()).await;

        // Node B
        p2p_config.bootstrap_nodes = node_a.multiaddrs();
        let mut node_b = build_service_from_config(p2p_config.clone()).await;

        let (tx_test_end, mut rx_test_end) = mpsc::channel::<bool>(1);

        let mut request_sent = false;

        loop {
            tokio::select! {
                message_sent = rx_test_end.recv() => {
                    // we received a signal to end the test
                    assert!(message_sent.unwrap(), "Received incorrect or missing message");
                    break;
                }
                node_a_event = node_a.next_event() => {
                    if let Some(FuelP2PEvent::PeerInfoUpdated { peer_id, block_height: _ }) = node_a_event {
                        if node_a.peer_manager.get_peer_info(&peer_id).is_some() {
                            // 0. verifies that we've got at least a single peer address to request message from
                            if !request_sent {
                                request_sent = true;

                                match request_msg.clone() {
                                    RequestMessage::SealedHeaders(range) => {
                                        let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseSender::SealedHeaders(tx_orchestrator)).is_ok());
                                        let tx_test_end = tx_test_end.clone();

                                        tokio::spawn(async move {
                                            let response_message = rx_orchestrator.await;

                                            let expected = arbitrary_headers_for_range(range.clone());

                                            if let Ok((_, Ok(sealed_headers))) = response_message {
                                                let check = expected.iter().zip(sealed_headers.unwrap().iter()).all(|(a, b)| eq_except_metadata(a, b));
                                                let _ = tx_test_end.send(check).await;
                                            } else {
                                                tracing::error!("Orchestrator failed to receive a message: {:?}", response_message);
                                                let _ = tx_test_end.send(false).await;
                                            }
                                        });
                                    }
                                    RequestMessage::Transactions(_range) => {
                                        let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseSender::Transactions(tx_orchestrator)).is_ok());
                                        let tx_test_end = tx_test_end.clone();

                                        tokio::spawn(async move {
                                            let response_message = rx_orchestrator.await;

                                            if let Ok((_, Ok(Some(transactions)))) = response_message {
                                                let check = transactions.len() == 1 && transactions[0].0.len() == 5;
                                                let _ = tx_test_end.send(check).await;
                                            } else {
                                                tracing::error!("Orchestrator failed to receive a message: {:?}", response_message);
                                                let _ = tx_test_end.send(false).await;
                                            }
                                        });
                                    }
                                }
                            }
                        }
                    }

                    tracing::info!("Node A Event: {:?}", node_a_event);
                },
                node_b_event = node_b.next_event() => {
                    // 2. Node B receives the RequestMessage from Node A initiated by the NetworkOrchestrator
                    if let Some(FuelP2PEvent::InboundRequestMessage{ request_id, request_message: received_request_message }) = &node_b_event {
                        match received_request_message {
                            RequestMessage::SealedHeaders(range) => {
                                let sealed_headers: Vec<_> = arbitrary_headers_for_range(range.clone());

                                let _ = node_b.send_response_msg(*request_id, ResponseMessage::SealedHeaders(Some(sealed_headers)));
                            }
                            RequestMessage::Transactions(_) => {
                                let txs = (0..5).map(|_| Transaction::default_test_tx()).collect();
                                let transactions = vec![Transactions(txs)];
                                let _ = node_b.send_response_msg(*request_id, ResponseMessage::Transactions(Some(transactions)));
                            }
                        }
                    }

                    tracing::info!("Node B Event: {:?}", node_b_event);
                }
            };
        }
    }