<commit_before>package node

import (
	"bytes"
	"encoding/json"
	"fmt"
	"github.com/c2stack/c2g/c2"
	"github.com/c2stack/c2g/meta"
	"io"
)

type Subscriber interface {
	Subscribe(sub *Subscription) error
}

type SubscriptionManager struct {
	subscriptions map[string]*Subscription
	factory       Subscriber
	Send          chan *SubscriptionMessage
	rdr           io.Reader
	wtr           io.Writer
	decoder       func(r io.Reader, conn *SubscriptionManager) error
	encoder       func(w io.Writer, msg *SubscriptionMessage) error
	lastErr       error
}

func NewSubscriptionManager(factory Subscriber, r io.Reader, w io.Writer) *SubscriptionManager {
	return &SubscriptionManager{
		rdr:           r,
		wtr:           w,
		factory:       factory,
		subscriptions: make(map[string]*Subscription),
		Send:          make(chan *SubscriptionMessage),
		decoder:       DecodeSubscriptionStream,
		encoder:       EncodeSubscriptionStream,
	}
}

func (self *SubscriptionManager) Run() error {
	go func() {
		self.checkErr(self.decoder(self.rdr, self))
		//self.Close()
	}()
	for {
		msg, open := <-self.Send
		if !open {
			break
		}
		if err := self.encoder(self.wtr, msg); err != nil {
			// This is fatal error sending, not err in application layer so
			// we should abort all further communication
			return err
		}
	}
	return self.lastErr
}

func (self *SubscriptionManager) checkErr(err error) {
	if err != nil && self.lastErr == nil {
		self.lastErr = err
	}
}

func (self *SubscriptionManager) Close() error {
	self.lastErr = nil
	for _, sub := range self.subscriptions {
		self.checkErr(sub.Close())
	}
	self.subscriptions = nil
	close(self.Send)
	return self.lastErr
}

// Subscribe to notifcation event
// SEND
// =============
// + foo job/states
//
// RECEIVE
// =============
// M foo some/path 51
// {"id":"5723adaa3684664e2b1936e9","state":"running"}
// M foo some/path 53
// {"id":"5723adaa3684664e2b1936e9","state":"succeeded"}
//
// Unsubscribe to notification event
// SEND
// =============
// - foo job/states
//
// *No response expected*
//
// Unsubscribe to all notification events for a context
// SEND
// =============
// - foo
//
// *No response expected*

type Subscription struct {
	Path         string
	Notification *meta.Notification
	Closer       NotifyCloser
	id           string
	group        string
	send         chan<- *SubscriptionMessage
}

func (self *Subscription) Notify(notification *meta.Notification, path *Path, n Node) {
	var buf bytes.Buffer
	json := NewJsonWriter(&buf).Node()
	sel := NewBrowser2(self.Notification, n).Root()
	err := sel.InsertInto(json).LastErr
	if err != nil {
		panic(err.Error())
	}
	self.send <- &SubscriptionMessage{
		Group:   self.group,
		Path:    path.StringNoModule(),
		Type:    "notify",
		Payload: buf.Bytes(),
	}
}

func (self *Subscription) Close() error {
	if self.Closer != nil {
		return self.Closer()
	}
	return nil
}

func EncodeSubscriptionStream(w io.Writer, msg *SubscriptionMessage) (err error) {
	return json.NewEncoder(w).Encode(msg)
}

type SubscriptionMessage struct {
	Path    string `json:"path"`
	Type    string `json:"type"`
	Group   string `json:"group,omitempty"`
	Payload []byte `json:"payload,omitempty"`
}

// This assumes each message will
//{
//  "op" : "+|-",
//  "context" : "x",
//  "path" : "y"
//}<commit_msg>allow notification payloads to be nil<commit_after>package node

import (
	"bytes"
	"encoding/json"
	"fmt"
	"github.com/c2stack/c2g/c2"
	"github.com/c2stack/c2g/meta"
	"io"
)

type Subscriber interface {
	Subscribe(sub *Subscription) error
}

type SubscriptionManager struct {
	subscriptions map[string]*Subscription
	factory       Subscriber
	Send          chan *SubscriptionMessage
	rdr           io.Reader
	wtr           io.Writer
	decoder       func(r io.Reader, conn *SubscriptionManager) error
	encoder       func(w io.Writer, msg *SubscriptionMessage) error
	lastErr       error
}

func NewSubscriptionManager(factory Subscriber, r io.Reader, w io.Writer) *SubscriptionManager {
	return &SubscriptionManager{
		rdr:           r,
		wtr:           w,
		factory:       factory,
		subscriptions: make(map[string]*Subscription),
		Send:          make(chan *SubscriptionMessage),
		decoder:       DecodeSubscriptionStream,
		encoder:       EncodeSubscriptionStream,
	}
}

func (self *SubscriptionManager) Run() error {
	go func() {
		self.checkErr(self.decoder(self.rdr, self))
		//self.Close()
	}()
	for {
		msg, open := <-self.Send
		if !open {
			break
		}
		if err := self.encoder(self.wtr, msg); err != nil {
			// This is fatal error sending, not err in application layer so
			// we should abort all further communication
			return err
		}
	}
	return self.lastErr
}

func (self *SubscriptionManager) checkErr(err error) {
	if err != nil && self.lastErr == nil {
		self.lastErr = err
	}
}

func (self *SubscriptionManager) Close() error {
	self.lastErr = nil
	for _, sub := range self.subscriptions {
		self.checkErr(sub.Close())
	}
	self.subscriptions = nil
	close(self.Send)
	return self.lastErr
}

// Subscribe to notifcation event
// SEND
// =============
// + foo job/states
//
// RECEIVE
// =============
// M foo some/path 51
// {"id":"5723adaa3684664e2b1936e9","state":"running"}
// M foo some/path 53
// {"id":"5723adaa3684664e2b1936e9","state":"succeeded"}
//
// Unsubscribe to notification event
// SEND
// =============
// - foo job/states
//
// *No response expected*
//
// Unsubscribe to all notification events for a context
// SEND
// =============
// - foo
//
// *No response expected*

type Subscription struct {
	Path         string
	Notification *meta.Notification
	Closer       NotifyCloser
	id           string
	group        string
	send         chan<- *SubscriptionMessage
}

func (self *Subscription) Notify(notification *meta.Notification, path *Path, n Node) {
	var payload []byte
	if n != nil {
		var buf bytes.Buffer
		json := NewJsonWriter(&buf).Node()
		sel := NewBrowser2(self.Notification, n).Root()
		err := sel.InsertInto(json).LastErr
		if err != nil {
			panic(err.Error())
		}
		payload = buf.Bytes()
	}
	self.send <- &SubscriptionMessage{
		Group:   self.group,
		Path:    path.StringNoModule(),
		Type:    "notify",
		Payload: payload,
	}
}

func (self *Subscription) Close() error {
	if self.Closer != nil {
		return self.Closer()
	}
	return nil
}

func EncodeSubscriptionStream(w io.Writer, msg *SubscriptionMessage) (err error) {
	return json.NewEncoder(w).Encode(msg)
}

type SubscriptionMessage struct {
	Path    string `json:"path"`
	Type    string `json:"type"`
	Group   string `json:"group,omitempty"`
	Payload []byte `json:"payload,omitempty"`
}

// This assumes each message will
//{
//  "op" : "+|-",
//  "context" : "x",
//  "path" : "y"
//}
{'fraction_non_alphanumeric': 0.1462214935161723, 'fraction_numerical': 0.01311670889849456, 'mean_word_length': 3.183291770573566, 'pattern_counts': {'":': 8, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6913856', 'n_tokens_mistral': 2307, 'n_tokens_neox': 2203, 'n_words': 567}