聊天室广播mod

asset/tmpl.html

@@ -68,7 +68,7 @@
                     function(resp){
                         if(resp !== null){
                             if(resp.ok) {
-                                $('#content').append('<p>'+resp.info+'</p>');
+                                //$('#content').append('<p>'+resp.info+'</p>'); 自己发的内容通过广播接收
                                 $('#line_recv').val('');
                             } else {
                                 $('#content').append('<p class="error">'+resp.info+'</p>');
@@ -111,6 +111,14 @@
             return false;
         });
 
+        //自动更新
+        (function poll(){
+            $.ajax({ url: "<?php echo get_baseurl() ?>:8421/sub", success: function(data){
+                $('#content').append('<p>'+data+'</p>');
+                scrollBottom();
+            }, complete: poll, timeout: 30000 });
+        })();
+
         //默认滚到底部
         scrollBottom();
     });

broadcast_svc/broadcast.go

@@ -0,0 +1,68 @@
+// broadcast
+package main
+
+import (
+	"sync"
+)
+
+type Broadcast struct { //各种地方不知道用指针还是怎么样
+	writeBuffer chan interface{}
+	listener    []*Receiver
+	lock        *sync.Mutex
+}
+
+type Receiver struct {
+	C    chan interface{}
+	Free bool
+}
+
+func NewBroadcast() (b *Broadcast) {
+	writeBuffer := make(chan interface{})
+	listener := []*Receiver{}
+
+	go func() { //分发过程
+		for {
+			news := <-b.writeBuffer
+			for k, _ := range b.listener {
+				if b.listener[k].Free { //已关闭的不投递消息
+					continue
+				}
+				b.listener[k].C <- news
+			}
+		}
+
+	}()
+	return &Broadcast{
+		writeBuffer: writeBuffer,
+		listener:    listener,
+		lock:        new(sync.Mutex),
+	}
+}
+
+func (b *Broadcast) Pub(news interface{}) {
+	go func() {
+		b.writeBuffer <- news
+	}()
+}
+
+func (b *Broadcast) Sub() (listener *Receiver) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	for k, _ := range b.listener {
+		if b.listener[k].Free {
+			b.listener[k].Free = false
+			return b.listener[k]
+		}
+	}
+	listener = &Receiver{C: make(chan interface{})}
+	b.listener = append(b.listener, listener)
+	return listener
+}
+
+func (r Receiver) Read() <-chan interface{} {
+	return r.C
+}
+
+func (r *Receiver) Close() {
+	r.Free = true
+}

broadcast_svc/main.go

@@ -0,0 +1,55 @@
+// broadcast_svc project main.go
+package main
+
+import (
+	"log"
+	"net/http"
+	"strings"
+)
+
+var B *Broadcast
+
+func main() {
+	B = NewBroadcast()
+
+	http.HandleFunc("/sub", onSub)
+	http.HandleFunc("/pub", onPub)
+	err := http.ListenAndServe(":8421", nil)
+	if err != nil {
+		log.Fatal("ListenAndServe: ", err)
+	}
+}
+
+func onSub(w http.ResponseWriter, req *http.Request) {
+	l := B.Sub()
+	defer l.Close()
+	cn := w.(http.CloseNotifier)
+
+	log.Println("Client", req.RemoteAddr, "sub the channel")
+
+	for {
+		select {
+		case msg := <-l.Read():
+			w.Header().Set("Access-Control-Allow-Origin", "*") //跨域问题
+			w.Write([]byte(msg.(string)))
+			log.Println("Msg sent to client", req.RemoteAddr)
+			return
+		case <-cn.CloseNotify():
+			log.Println("Client", req.RemoteAddr, "disconnected")
+			break
+		}
+	}
+}
+
+func onPub(w http.ResponseWriter, req *http.Request) {
+
+	if !strings.HasPrefix(req.RemoteAddr, "127.0.0.1") {
+		w.Write([]byte("Broadcast from localhost ONLY!"))
+		return
+	}
+
+	req.ParseForm()
+	news := req.Form.Get("line")
+	B.Pub(news)
+	log.Println("Msg sent to channel:", news)
+}

index.php

@@ -66,6 +66,13 @@ function m_save() {
             $time = date('[Y/m/d H:i:s] ');
             $log_line = "{$time}{$_POST['log_line']}\n"; //追加日期格式
             if(write_line($log_line)) {
+                //广播过程
+                $ch = curl_init(get_baseurl().":8421/pub?line=".urlencode(htmlspecialchars($log_line)));
+                curl_setopt_array($ch, array(
+                    CURLOPT_RETURNTRANSFER => true
+                ));
+                curl_exec($ch);
+
                 die(json_encode(array('ok' => true, 'info' => htmlspecialchars($log_line))));
             } else {
                 die(json_encode(array('ok' => false, 'info' => $time.'Write File Error!')));