[TypeScript] P2P(peer-to-peer) 구현

1. P2P 개념

2. P2P 코드 구현

---

1. P2P 개념

[그림 1] Server-Based와 P2P

P2P는 동등 계층간 통신망이라고도 하며, 비교적 소수의 서버에 집중하기보다는 망 구성에 참여하는 기계들의 계산과 대역폭 성능에 의존하여 구성되는 통신망이다. P2P 통신망은 일반적으로 노드들을 규모가 큰 애드혹으로 서로 연결하는 경우 이용된다.

 

순수 P2P 파일 전송 네트워크는 클라이언트나 서버라는 개념 없이, 오로지 동등한 계층 노드들(peer nodes)이 클라이언트와 서버 역할을 동시에 네트워크 위에서 하게 된다.

 

 

2. P2P 코드 구현

/* index.ts */

import { BlockChain } from '@core/index'
import { P2PServer } from './src/serve/p2p'
import peers from './peer.json'
import express from 'express'

enum MessageType {
    latest_block = 0,
    all_block = 1,
    received_chain = 2,
}

interface Message {
    type: MessageType
    payload: any
}

const app = express()
const bc = new BlockChain()
const ws = new P2PServer()

app.use(express.json())

app.use((req, res, next) => {
    const baseAuth: string = (req.headers.authorization || '').split(' ')[1]
    if (baseAuth === '') return res.status(401).send()
    const [userid, userpw] = Buffer.from(baseAuth, 'base64').toString().split(':')
    if (userid !== 'jenny' || userpw !== '1234') return res.status(401).send()
    next()
})

app.get('/', (req, res) => {
    res.send('jenny chain')
})

app.get('/chains', (req, res) => {
    res.json(ws.getChain())
})

app.post('/mineBlock', (req, res) => {
    const { data } = req.body
    const newBlock = ws.addBlock(data)
    if (newBlock.isError) return res.status(500).json(newBlock.error)
    const msg: Message = {
        type: MessageType.latest_block,
        payload: {},
    }
    ws.broadcast(msg)
    res.json(newBlock.value)
})

app.post('/addToPeer', (req, res) => {
    const { peer } = req.body
    ws.connectionToPeer(peer)
    res.json(peer)
})

app.post('/addPeers', (req, res) => {
    peers.forEach((peer) => {
        ws.connectionToPeer(peer)
    })
})

app.get('/peers', (req, res) => {
    const sockets = ws.getSockets().map((s: any) => {
        return s._socket.remoteAddress + ':' + s._socket.remotePort
    })
    res.json(sockets)
})

app.listen(3000, () => {
    console.log('server start 3000')
    ws.listen()
})

위 코드는 index.ts 파일의 내용이다. 

 

app.use((req, res, next) => {}) 라우터에서는 사용자 인증을 거친다. 

[그림 2] 사용자 id, pw 검증

위 사진처럼 get 방식으로 id에 jenny, pw에 1234를 넣어서 요청을 보내면 status 코드가 200이 뜬다.

status 코드 401은 Unauthorized로 인증되지 않은 상태에 대한 내용이다.

 

/mineBlock 은 body에 data를 담아서 요청을 보내면 새로운 블록이 생성되도록 해주는 라우터이다.

 

/addToPeer 는 body에 peer 정보를 담아서 요청을 보내면 peer가 추가되도록 하는 라우터이다. 이는 peer를 일일이 추가해주어야 하기 때문에 번거롭다.

 

/addPeers 는 peer.json이라는 파일에 아래와 같이 작성해두면 서버를 켰을 때 자동으로 peer가 추가된다.

[
    "ws://123.123.123.123:7545", 
    "ws://111.111.111.111:7545", 
    "ws://222.222.222.222:7545"
]

 

 

이제 p2p.ts 파일에 대해 살펴보겠다.

/* p2p.ts */

import { WebSocket } from 'ws'
import { Chain } from '@core/blockchain/chain'

enum MessageType {
    latest_block = 0,
    all_block = 1,
    received_chain = 2,
}

interface Message {
    type: MessageType
    payload: any
}

export class P2PServer extends Chain {
    private sockets: WebSocket[]
    constructor() {
        super()
        this.sockets = []
    }

    getSockets() {
        return this.sockets
    }

    listen() {
        const server = new WebSocket.Server({ port: 7545 })
        server.on('connection', (socket) => {
            console.log(`-----------------> WebSocket Connection ${new Date()} <-----------------`)
            this.connectSocket(socket)
        })
    }

    connectionToPeer(newPeer: string) {
        const socket = new WebSocket(newPeer)
        socket.on('open', () => {
            this.connectSocket(socket)
        })
    }

    connectSocket(socket: WebSocket) {
        this.sockets.push(socket)
        this.messageHandler(socket)
        const data: Message = {
            type: MessageType.latest_block,
            payload: {},
        }
        this.errorHandler(socket)
        const send = P2PServer.send(socket)
        send(data)
    }

    messageHandler(socket: WebSocket) {
        const callback = (data: string) => {
            const result: Message = P2PServer.dataParse<Message>(data)
            const send = P2PServer.send(socket)
            switch (result.type) {
                case MessageType.latest_block: {
                    const message: Message = {
                        type: MessageType.all_block,
                        payload: [this.getLatestBlock()],
                    }
                    send(message)
                    break
                }
                case MessageType.all_block: {
                    const message: Message = {
                        type: MessageType.received_chain,
                        payload: this.getChain(),
                    }
                    const [receivedBlock] = result.payload
                    const isValid = this.addToChain(receivedBlock)
                    if (isValid.isError) send(message)
                    break
                }
                case MessageType.received_chain: {
                    const received_chain: IBlock[] = result.payload
                    this.handleChainResponse(received_chain)
                    console.log(received_chain)
                    break
                }
            }
        }
        socket.on('message', callback)
    }

    errorHandler(socket: WebSocket) {
        const close = () => {
            this.sockets.splice(this.sockets.indexOf(socket), 1)
        }
        socket.on('close', close)
        socket.on('error', close)
    }

    static send(_socket: WebSocket) {
        return (_data: Message) => {
            _socket.send(JSON.stringify(_data))
        }
    }

    broadcast(message: Message): void {
        this.sockets.forEach((socket) => P2PServer.send(socket)(message))
    }

    handleChainResponse(received_chain: IBlock[]): Failable<Message | undefined, string> {
        const isValidChain = this.isValidChain(received_chain)
        if (isValidChain.isError) return { isError: true, error: isValidChain.error }
        const isValid = this.replaceChain(received_chain)
        if (isValid.isError) return { isError: true, error: isValid.error }
        this.blockchain = received_chain
        const message: Message = {
            type: MessageType.received_chain,
            payload: received_chain,
        }
        this.broadcast(message)
        return { isError: false, value: undefined }
    }

    static dataParse<T>(_data: string): T {
        return JSON.parse(Buffer.from(_data).toString())
    }
}

websocket 서버의 포트 번호는 7545이다. 

 

server가 connection이 맺어졌을 때 connectSocket() 메소드가 실행된다.

 

messageHandler()는 MessageType에 따라 switch문으로 처리된다. MessageType은 가장 최신의 블록을 가져올 때, 모든 블록을 가져올 때, 그리고 다른 사람의 chain으로 변경되었을 때에 따라 정해진다.

 

errorHandler()는 소켓 연결이 끊어지거나 에러가 발생했을 때 this.sockets에서 해당 소켓을 제거한다.

 

broadcast()는 연결되어 있는 모든 peer에게 메시지를 보내는 메소드이다.

 

handleChainResponse()는 내 chain이 다른 사람의 chain으로 교체되었을 때 broadcast로 peer에게 메시지를 보내는 메소드이다.