     1	/*
     2	  My Block Chain: P2P Network module
     3	*/
     4	package P2P
     5	
     6	import (
     7		"encoding/json"
     8		"errors"
     9		"fmt"
    10		"net"
    11		"strconv"
    12		"time"
    13	)
    14	
    15	const (
    16		CMD_NEWBLOCK    = 1
    17		CMD_ADDSRV      = 2
    18		CMD_DELSRV      = 3
    19		CMD_SENDBLOCK   = 4
    20		CMD_MININGBLOCK = 5
    21		CMD_MODIFYDATA  = 6
    22	
    23		debug_mode = false
    24	)
    25	
    26	// サーバ管理の構造体
    27	type Node struct {
    28		Host    string   `json:"host" form:"host" query:"host"`
    29		ApiPort uint16   `json:"api_port" form :"api_port" query:"api_port"`
    30		P2PPort uint16   `json:"p2p_port" form :"p2p_port" query:"p2p_port"`
    31		Self    bool     `json:"-"`
    32		Conn    net.Conn `json:"-"`
    33	}
    34	
    35	// ネットワーク接続
    36	func (node *Node) connect() {
    37		target := node.Host + ":" + strconv.Itoa(int(node.P2PPort))
    38	
    39		if debug_mode {
    40			fmt.Println("target = ", target)
    41		}
    42	
    43		conn, err := net.Dial("udp", target)
    44		if err != nil {
    45			fmt.Println("failed to connect ", target, err)
    46			node.Conn = nil
    47		} else {
    48			fmt.Println(target, "connected.")
    49			node.Conn = conn
    50		}
    51	}
    52	
    53	// ネットワーク切断
    54	func (node *Node) disconnect() {
    55		if node.Conn != nil {
    56			node.Conn.Close()
    57		}
    58	}
    59	
    60	// メッセージ送信
    61	func (node *Node) Send(msg []byte) error {
    62		fmt.Println("Send to ", node.me(), ":", string(msg), len(msg))
    63	
    64		err := error(nil)
    65		if node.Conn != nil {
    66			n, err := node.Conn.Write(msg)
    67			if err != nil {
    68				fmt.Println("Write error:", n, err)
    69			}
    70		} else {
    71			err = errors.New("Not connected:" + node.me())
    72			fmt.Println("Not connected:", node.me())
    73		}
    74	
    75		return err
    76	}
    77	
    78	// サーバのアドレス情報の組み立て
    79	func (node *Node) me() string {
    80		return node.Host + ":" + strconv.FormatInt(int64(node.P2PPort), 10)
    81	}
    82	
    83	type act_fn func([]byte) error
    84	
    85	// 自身のアドレス情報を返す
    86	func (p2p *P2PNetwork) Self() string {
    87		for _, n := range p2p.nodes {
    88			if n.Self {
    89				return n.me()
    90			}
    91		}
    92		return ""
    93	}
    94	
    95	// P2P通信のサーバ処理
    96	func (p2p *P2PNetwork) p2p_srv(host string, port uint16) {
    97		fmt.Println("Start p2p server", host, port)
    98	
    99		udpAddr := &net.UDPAddr{
   100			IP:   net.ParseIP(host),
   101			Port: int(port),
   102		}
   103		updLn, err := net.ListenUDP("udp", udpAddr)
   104		if err != nil {
   105			fmt.Println("listen error", err)
   106			return
   107		}
   108	
   109		for {
   110			buf := make([]byte, 1024)
   111			if debug_mode {
   112				fmt.Println("call updLn.ReadFromUDP")
   113			}
   114			n, addr, err := updLn.ReadFromUDP(buf)
   115			if debug_mode {
   116				fmt.Println("read", n, err)
   117			}
   118			if err == nil {
   119				go func() {
   120					if debug_mode {
   121						fmt.Println("recieve function")
   122						fmt.Println(addr)
   123						fmt.Println(string(buf))
   124					}
   125					cmd := int(buf[0])
   126					if debug_mode {
   127						fmt.Println(cmd)
   128					}
   129					msg := buf[1:n]
   130					if cmd < len(p2p.actions) {
   131						fmt.Println("Do Action")
   132						f := p2p.actions[cmd]
   133						if f != nil {
   134							err := f(msg)
   135							if err != nil {
   136								fmt.Println(err)
   137							}
   138						}
   139					} else {
   140						fmt.Println("No Action")
   141					}
   142	
   143				}()
   144			}
   145		}
   146	}
   147	
   148	// P2Pネットワーク管理構造体
   149	type P2PNetwork struct {
   150		nodes   []*Node
   151		actions []act_fn
   152	}
   153	
   154	// P2Pネットワークにサーバを追加
   155	func (p2p *P2PNetwork) Add(node *Node) (int, error) {
   156	
   157		fmt.Println("P2PNetwork.Add")
   158	
   159		fmt.Println("add node:", node)
   160	
   161		// 他のサーバにも追加リクエストを飛ばす
   162		bytes, _ := json.Marshal(node)
   163		p2p.Broadcast(CMD_ADDSRV, bytes, false)
   164	
   165		// 通信準備
   166		node.connect()
   167	
   168		// 追加されたサーバに他のサーバ情報を送る
   169		for _, n := range p2p.nodes {
   170			b, _ := json.Marshal(n)
   171			s_msg := append([]byte{byte(CMD_ADDSRV)}, b...)
   172			node.Send(s_msg)
   173			time.Sleep(1 * time.Second / 2)
   174		}
   175	
   176		// サーバリストに追加
   177		p2p.nodes = append(p2p.nodes, node)
   178	
   179		return 0, nil
   180	}
   181	
   182	// サーバ情報の検索
   183	func (p2p *P2PNetwork) Search(host string, p2p_port uint16) *Node {
   184	
   185		fmt.Println("Search:", host, p2p_port)
   186	
   187		for _, node := range p2p.nodes {
   188			if node.Host == host && node.P2PPort == p2p_port {
   189				return node
   190			}
   191		}
   192	
   193		return nil
   194	}
   195	
   196	// P2Pネットワークに接続しているサーバ一覧を取得
   197	func (p2p *P2PNetwork) List() []*Node {
   198		for _, node := range p2p.nodes {
   199			fmt.Println(node)
   200		}
   201		return p2p.nodes
   202	}
   203	
   204	// P2Pネットワークに接続しているサーバにメッセージ送信
   205	func (p2p *P2PNetwork) Broadcast(cmd int, msg []byte, self bool) {
   206	
   207		fmt.Println("Broadcast:", cmd, string(msg))
   208		/*
   209		   メッセージ送信時は、cmd + msg で送る。
   210		   cmd は 1バイトとする。
   211		*/
   212		s_msg := append([]byte{byte(cmd)}, msg...)
   213		if debug_mode {
   214			fmt.Println(s_msg)
   215		}
   216	
   217		for _, node := range p2p.nodes {
   218			if debug_mode {
   219				fmt.Println(node)
   220			}
   221			if self == false && node.Self {
   222				fmt.Println("not send")
   223				continue
   224			} else {
   225				if err := node.Send(s_msg); err != nil {
   226					fmt.Println("send error:", node, err)
   227				}
   228			}
   229			time.Sleep(1 * time.Second / 2)
   230		}
   231	}
   232	
   233	// メッセージをいずれかのサーバに送信
   234	func (p2p *P2PNetwork) SendOne(cmd int, msg []byte) {
   235		fmt.Println("SendOne:", cmd, string(msg))
   236	
   237		s_msg := append([]byte{byte(cmd)}, msg...)
   238		if debug_mode {
   239			fmt.Println(s_msg)
   240		}
   241	
   242		for _, node := range p2p.nodes {
   243			if debug_mode {
   244				fmt.Println(node)
   245			}
   246			if node.Self {
   247				fmt.Println("not send")
   248				continue
   249			} else {
   250				err := node.Send(s_msg)
   251				if err == nil {
   252					break
   253				}
   254			}
   255			time.Sleep(1 * time.Second / 2)
   256		}
   257	}
   258	
   259	// アクションとアクションハンドラの紐付け登録
   260	func (p2p *P2PNetwork) SetAction(cmd int, handler act_fn) *act_fn {
   261	
   262		fn := p2p.actions[cmd]
   263		p2p.actions[cmd] = handler
   264		return &fn
   265	}
   266	
   267	// P2P ネットワークの初期化処理
   268	func (p2p *P2PNetwork) Init(host string, api_port uint16, p2p_port uint16) (*P2PNetwork, error) {
   269	
   270		fmt.Println("P2P_init")
   271		p2p.nodes = make([]*Node, 0)
   272		p2p.actions = make([]act_fn, 20)
   273	
   274		// 自ノードの管理構造を初期化
   275		node := new(Node)
   276		node.Host = host
   277		node.ApiPort = api_port
   278		node.P2PPort = p2p_port
   279		node.Self = true
   280	
   281		// 自ノードの通信路開設
   282		node.connect()
   283	
   284		// サーバリストに自ノードを追加
   285		p2p.nodes = append(p2p.nodes, node)
   286	
   287		// サーバ初期化
   288		go p2p.p2p_srv(host, p2p_port)
   289	
   290		if debug_mode {
   291			fmt.Println(p2p)
   292		}
   293	
   294		return p2p, nil
   295	}
   296	
   297	// サーバ追加アクション
   298	func (p2p *P2PNetwork) AddSrv(msg []byte) error {
   299		fmt.Println("add server action")
   300	
   301		node := new(Node)
   302	
   303		if debug_mode {
   304			fmt.Println(msg)
   305			fmt.Println(string(msg))
   306		}
   307	
   308		err := json.Unmarshal(msg, node)
   309		if err != nil {
   310			fmt.Println("json.Unmarshal failed")
   311			return err
   312		}
   313		fmt.Println("node:", node)
   314	
   315		/*
   316		   サーバリストに追加して、通信路を接続する。
   317		   追加するとき、Selfはfalseにすること
   318		*/
   319		node.Self = false
   320		node.connect()
   321		p2p.nodes = append(p2p.nodes, node)
   322	
   323		return nil
   324	}
   325	
