1package net
 2
 3import (
 4	"os"
 5	"fmt"
 6	"encoding/json"
 7
 8	shell "github.com/ipfs/go-ipfs-api"
 9	"github.com/vocdoni/go-dvote/batch"
10	"github.com/vocdoni/go-dvote/types"
11)
12
13type PubSubHandle struct {
14	topic string
15	subscription *shell.PubSubSubscription
16}
17
18func PsSubscribe(topic string) *shell.PubSubSubscription {
19	sh := shell.NewShell("localhost:5001")
20	sub, err := sh.PubSubSubscribe(topic)
21	if err != nil {
22		fmt.Fprintf(os.Stderr, "error: %s", err)
23		os.Exit(1)
24	}
25	return sub
26}
27
28func PsPublish(topic, data string) error {
29	sh := shell.NewShell("localhost:5001")
30	err := sh.PubSubPublish(topic, data)
31	if err != nil {
32		return err
33	}
34	return nil
35}
36
37func (p *PubSubHandle) Init(topic string) error {
38	p.topic = topic
39	p.subscription = PsSubscribe(p.topic)
40	return nil
41}
42
43func (p *PubSubHandle) Listen() error {
44	var msg *shell.Message
45	var err error
46	for {
47		msg, err = p.subscription.Next()
48		if err != nil {
49			fmt.Fprintf(os.Stderr, "recieve error: %s", err)
50			return err
51		}
52
53		payload := msg.Data
54
55		var e types.Envelope
56		var b types.Ballot
57
58		err = json.Unmarshal(payload, &e)
59		if err != nil {
60			return err
61		}
62
63		err = json.Unmarshal(e.Ballot, &b)
64		if err != nil {
65			return err
66		}
67
68		err = batch.Add(b)
69		if err != nil {
70			return err
71		}
72
73		fmt.Println("Got > " + string(payload))
74	}
75}
76
77func (p *PubSubHandle) Send(data string) error {
78	return PsPublish(p.topic, data)
79}