async _handleProtocolStream({ stream, connection }) {
		try {
			// Get peer ID
			const peerId = connection.remotePeer.toString();
			logT('p2p', 'New protocol stream from peer', peerId);
			
			// Read the message from the stream
			const chunks = [];
			try {
				for await (const chunk of stream.source) {
					chunks.push(chunk.subarray());
				}
			} catch (err) {
				logTE('p2p', `Error reading from stream from peer ${peerId}:`, err);
				await stream.abort(err);
				return;
			}
			
			// Parse the message
			const message = JSON.parse(Buffer.concat(chunks).toString());
			const topic = message.topic;
			
			if (!topic) {
				logTW('p2p', 'Received message without topic', message);
				return;
			}
			
			// For init topic, we always process it
			const isInitTopic = topic === `${this.protocol}/init`;
			
			// For other topics, verify protocol compatibility
			if (!isInitTopic && this.peers.has(peerId) && !this.peers.get(peerId)[this.protocolName]) {
				logTW('p2p', `Ignoring message from unverified peer ${peerId} on topic ${topic}`);
				return;
			}

			const messageData = message.data || message;
			
			// Create respond function for request-response pattern
			const respond = async (responseData) => {
				try {
					// Create response message
					const response = {
						data: responseData,
						id: message.id,
						isResponse: true,
						topic
					};
					
					// Convert to buffer
					const responseBuffer = Buffer.from(JSON.stringify(response));
					
					// Send response
					await this._sendProtocolStreamData(peerId, responseBuffer);
					logT('p2p', `Sent direct response to ${peerId} for topic ${topic}`);
				} catch (err) {
					logTE('p2p', 'Error sending direct response', err);
				}
			};
			
			// Check if this is a response to a request
			if (message.id && message.isResponse && this.responseHandlers.has(message.id)) {
				const responseHandler = this.responseHandlers.get(message.id);
				responseHandler(messageData, peerId);
				
				// Remove one-time handlers
				if (!responseHandler.permanent) {
					this.responseHandlers.delete(message.id);
				}
				return;
			}
			
			// Handle topic-specific messages
			if (this.topicHandlers.has(topic)) {
				const topicHandler = this.topicHandlers.get(topic).handler;
				
				// For request-response pattern, include the ability to reply
				topicHandler(messageData, peerId, respond);
			} else {
				logTW('p2p', `No handler for topic ${topic}`);
			}
		} catch (err) {
			logTE('p2p', 'Error handling protocol stream', err);
		} finally {
			await stream.close();
		}
	}
Handle incoming protocol messages
@param {Event} evt - Message event