openidec

net.go (8406B)

---

 // Network operations: fetch, post/get message from point
 // Check node extensions.
 
 package ii
 
 import (
 	"bufio"
 	"encoding/base64"
 	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
 
 	//	"net/smtp"
 	"net/url"
 	"strings"
 	"sync"
 )
 
 // Node object. Use Connect to create it.
 // Host: url node
 // Features: extensions map
 // Force: force sync even last message is not new
 type Node struct {
 	Host     string
 	Features map[string]bool
 	Force    bool
 }
 
 // utility function to make get request and call fn
 // for every line. Stops on EOF or fn return false.
 func http_req_lines(url string, fn func(string) bool) error {
 	resp, err := http.Get(url)
 	if err != nil {
 		return err
 	}
 	defer resp.Body.Close()
 	reader := bufio.NewReader(resp.Body)
 	for {
 		line, err := reader.ReadString('\n')
 		if err != nil && err != io.EOF {
 			return err
 		}
 		line = strings.TrimSuffix(line, "\n")
 		if err == io.EOF {
 			if line != "" { /* node do not send final \n */
 				fn(line)
 			}
 			break
 		}
 		if !fn(line) {
 			break
 		}
 	}
 	return nil
 }
 
 // short variant of http_get_lines. Read one line and
 // interpret it as message id. Return it.
 func http_get_id(url string) (string, error) {
 	res := ""
 	if err := http_req_lines(url, func(line string) bool {
 		if strings.Contains(line, ".") {
 			return true
 		}
 		res += line
 		return true
 	}); err != nil {
 		return "", err
 	}
 	return res, nil
 }
 
 // Fetcher internal goroutine.
 // DB: db to write
 // Echo: echo to fetch
 // wait: sync for Fetch master to detect finishing of work
 // cond: used for wake-up new goroutines
 // Can work in different modes.
 // If limit > 0, just fetch last [limit] messages (-limit:limit slice)
 // if limit < 0, use adaptive mode, probe (-(2*n)* limit:1) messages
 // untill find old message.
 // if node does not support u/e slices, than full sync performed
 // if node connection is not in Force mode, do not perform sync if not needed
 func (n *Node) Fetcher(db *DB, Echo string, limit int, wait *sync.WaitGroup, cond *sync.Cond) {
 	defer func() {
 		cond.L.Lock()
 		cond.Broadcast()
 		cond.L.Unlock()
 	}()
 	defer wait.Done()
 	if n.IsFeature("u/e") { /* fast path */
 		if !n.Force {
 			id, err := http_get_id(n.Host + "/u/e/" + Echo + "/-1:1")
 			if !IsMsgId(id) {
 				Info.Printf("%s: no valid MsgId", Echo)
 				return
 			}
 			if err == nil && db.Exists(id) != nil { /* no sync needed */
 				Info.Printf("%s: no sync needed", Echo)
 				return
 			}
 		}
 		if limit < 0 {
 			limit = -limit
 			try := 0
 			for { // adaptive
 				if try > 16 { /* fallback to old scheme */
 					limit = 0
 					break
 				}
 				id, err := http_get_id(fmt.Sprintf("%s/u/e/%s/%d:1",
 					n.Host, Echo, -limit))
 				if err != nil { /* fallback to old scheme */
 					limit = 0
 					break
 				}
 				if db.Exists(id) != nil {
 					break
 				}
 				try++
 				limit *= 2
 			}
 		}
 	} else {
 		limit = 0
 	}
 	req := fmt.Sprintf("%s/u/e/%s", n.Host, Echo)
 	if limit > 0 {
 		req = fmt.Sprintf("%s/%d:%d", req, -limit, limit)
 	}
 	Info.Printf("Get %s", req)
 	var res []string
 	if err := http_req_lines(req, func(line string) bool {
 		if strings.Contains(line, ".") {
 			return true
 		}
 		if db.Exists(line) == nil {
 			res = append(res, line)
 		}
 		return true
 	}); err != nil {
 		return
 	}
 
 	// FIXME: handle this error
 	_ = n.Store(db, res)
 }
 
 // Do not run more then MaxConnections goroutines in the same time
 var MaxConnections = 6
 
 // Send point message to node using GET method of /u/point scheme.
 // pauth: secret string. msg - raw message in plaintext
 // returns error
 func (n *Node) Send(pauth string, msg string) error {
 	msg = base64.URLEncoding.EncodeToString([]byte(msg))
 	//	msg = url.QueryEscape(msg)
 	req := fmt.Sprintf("%s/u/point/%s/%s", n.Host, pauth, msg)
 	resp, err := http.Get(req)
 	Trace.Printf("Get %s", req)
 	if err != nil {
 		return err
 	}
 	buf, err := ioutil.ReadAll(resp.Body)
 	if strings.HasPrefix(string(buf), "msg ok") {
 		Trace.Printf("Server responced msg ok")
 		return nil
 	} else if len(buf) > 0 {
 		err = errors.New(string(buf))
 	}
 	if err == nil {
 		err = errors.New("Server did not response with ok")
 	}
 	return err
 }
 
 // Send point message to node using POST method of /u/point scheme.
 // pauth: secret string. msg - raw message in plaintext
 // returns error
 func (n *Node) Post(pauth string, msg string) error {
 	msg = base64.StdEncoding.EncodeToString([]byte(msg))
 	// msg = url.QueryEscape(msg)
 	postData := url.Values{
 		"pauth": {pauth},
 		"tmsg":  {msg},
 	}
 	resp, err := http.PostForm(n.Host+"/u/point", postData)
 	Trace.Printf("Post %s/u/point", n.Host)
 	if err != nil {
 		return err
 	}
 	buf, err := ioutil.ReadAll(resp.Body)
 	if strings.HasPrefix(string(buf), "msg ok") {
 		Trace.Printf("Server responced msg ok")
 		return nil
 	} else if len(buf) > 0 {
 		err = errors.New(string(buf))
 	}
 	if err == nil {
 		err = errors.New("Server did not response with ok")
 	}
 	return err
 }
 
 // Return list.txt in []string if node supports it.
 // WARNING: Only echo names are returned! Each string is just echoarea.
 // Used for fetch all mode.
 func (n *Node) List() ([]string, error) {
 	var list []string
 	if !n.IsFeature("list.txt") {
 		return list, nil
 	}
 	if err := http_req_lines(n.Host+"/list.txt", func(line string) bool {
 		list = append(list, strings.Split(line, ":")[0])
 		return true
 	}); err != nil {
 		return list, err
 	}
 	return list, nil
 }
 
 // Fetch and write selected messages in db.
 // ids: selected message ids.
 // db: Database.
 // This function make /u/m request, decodes bundles, checks,
 // and write them to db (line by line).
 func (n *Node) Store(db *DB, ids []string) error {
 	req := ""
 	var nreq int
 	count := len(ids)
 	Trace.Printf("Get and store messages")
 	for i := 0; i < count; i++ {
 		req = req + "/" + string(ids[i])
 		nreq++
 		if nreq < 8 && i < count-1 {
 			continue
 		}
 		if err := http_req_lines(n.Host+"/u/m"+req, func(b string) bool {
 			m, e := DecodeBundle(b)
 			if e != nil {
 				Error.Printf("Can not decode message %s (%s)\n", b, e)
 				return true
 			}
 			if e := db.Store(m); e != nil {
 				Error.Printf("Can not write message %s (%s)\n", m.MsgId, e)
 			}
 			return true
 		}); err != nil {
 			return err
 		}
 		nreq = 0
 		req = ""
 	}
 	return nil
 }
 
 // This is Fetcher master function. It makes fetch from node
 // and run goroutines in parralel mode (one goroutine per echo).
 // Echolist: list with echoarea names. If list is empty,
 // function will try to get list via list.txt request.
 // limit: see Fetcher function. Describe fetching mode/limit.
 func (n *Node) Fetch(db *DB, Echolist []string, limit int) error {
 	if len(Echolist) == 0 {
 		Echolist, _ = n.List()
 	}
 	if Echolist == nil {
 		return nil
 	}
 	var wait sync.WaitGroup
 	cond := sync.NewCond(&sync.Mutex{})
 	num := 0
 	Info.Printf("Start fetcher(s) for %s", n.Host)
 	for _, v := range Echolist {
 		if !IsEcho(v) {
 			if strings.Trim(v, " ") != "" {
 				Trace.Printf("Skip echo: %s", v)
 			}
 			continue
 		}
 		wait.Add(1)
 		num += 1
 		if num >= MaxConnections { /* add per one */
 			cond.L.Lock()
 			Trace.Printf("Start fetcher for: %s", v)
 			go n.Fetcher(db, v, limit, &wait, cond)
 			Trace.Printf("Waiting free thread")
 			cond.Wait()
 			cond.L.Unlock()
 		} else {
 			Trace.Printf("Start fetcher for: %s", v)
 			go n.Fetcher(db, v, limit, &wait, cond)
 		}
 	}
 	Trace.Printf("Waiting thread(s)")
 	wait.Wait()
 	return nil
 }
 
 // Check if node has feature?
 // Features are gets while Connect call.
 func (n *Node) IsFeature(f string) bool {
 	_, ok := n.Features[f]
 	return ok
 }
 
 // Connect to node, get features and returns
 // pointer to Node object.
 func Connect(addr string) (*Node, error) {
 	var n Node
 	n.Host = strings.TrimSuffix(addr, "/")
 	n.Features = make(map[string]bool)
 	if err := http_req_lines(n.Host+"/x/features", func(line string) bool {
 		n.Features[line] = true
 		Trace.Printf("%s supports %s", n.Host, line)
 		return true
 	}); err != nil {
 		return nil, err
 	}
 	return &n, nil
 }
 
 /*
 // commented out routine to send e-mails ;)
 func SendMail(email string, login string, passwd string, server string) error {
 	aserv := strings.Split(server, ":")[0]
 	auth := smtp.PlainAuth("", login, passwd, aserv)
 	msg := "Hello!"
 	msg = "From: noreply@ii-go\n" +
 		"To: " + email + "\n" +
 		"Subject: Hello there\n\n" +
 		msg
 	err := smtp.SendMail(server, auth, "noreply@ii-go",[]string{email}, []byte(msg))
 	if err != nil {
 		Error.Printf("Can't send message to: %s", email)
 		return err
 	}
 	Info.Printf("Sent message to: %s", email)
 	return nil
 }
 */