func (s *State) gopush() {
	s.push.mut.Lock()
	t0 := time.Now()
	//queue cleanup
	defer func() {
		//measure time passed, ensure we wait at least Throttle time
		tdelta := time.Now().Sub(t0)
		if t := s.Throttle - tdelta; t > 0 {
			time.Sleep(t)
		}
		//push complete
		s.push.mut.Unlock()
		atomic.StoreUint32(&s.push.ing, 0)
		//if queued, auto-push again
		if atomic.CompareAndSwapUint32(&s.push.queued, 1, 0) {
			s.Push()
		}
	}()
	//calculate new json state
	l, hasLock := s.gostruct.(sync.Locker)
	if hasLock {
		l.Lock()
	}
	newBytes, err := json.Marshal(s.gostruct)
	if hasLock {
		l.Unlock()
	}
	if err != nil {
		log.Printf("velox: marshal failed: %s", err)
		return
	}
	//if changed, then calculate change set
	if !bytes.Equal(s.data.bytes, newBytes) {
		//calculate change set from last version
		ops, _ := jsonpatch.CreatePatch(s.data.bytes, newBytes)
		if len(s.data.bytes) > 0 && len(ops) > 0 {
			//changes! bump version
			s.data.mut.Lock()
			s.data.delta, _ = json.Marshal(ops)
			s.data.bytes = newBytes
			s.data.version++
			s.data.mut.Unlock()
		}
	}
	//send this new change to each subscriber
	s.connMut.Lock()
	for _, c := range s.conns {
		if c.version != s.data.version {
			go c.push()
		}
	}
	s.connMut.Unlock()
	//defered cleanup()
}
func (s *State) gopush() {
	s.push.mut.Lock()
	t0 := time.Now()
	//queue cleanup
	defer func() {
		//measure time passed, ensure we wait at least Throttle time
		tdelta := time.Now().Sub(t0)
		if t := s.Throttle - tdelta; t > 0 {
			time.Sleep(t)
		}
		//push complete
		s.push.mut.Unlock()
		atomic.StoreUint32(&s.push.ing, 0)
		//if queued, auto-push again
		if atomic.CompareAndSwapUint32(&s.push.queued, 1, 0) {
			s.Push()
		}
	}()
	//calculate new json state
	l, hasLock := s.gostruct.(sync.Locker)
	if hasLock {
		l.Lock()
	}
	newBytes, err := json.Marshal(s.gostruct)
	if hasLock {
		l.Unlock()
	}
	if err != nil {
		log.Printf("velox: marshal failed: %s", err)
		return
	}
	//if changed, then calculate change set
	if !bytes.Equal(s.data.bytes, newBytes) {
		//calculate change set from last version
		ops, _ := jsonpatch.CreatePatch(s.data.bytes, newBytes)
		if len(s.data.bytes) > 0 && len(ops) > 0 {
			//changes! bump version
			s.data.mut.Lock()
			s.data.delta, _ = json.Marshal(ops)
			s.data.bytes = newBytes
			s.data.version++
			s.data.mut.Unlock()
		}
	}
	//send this new change to each subscriber
	s.connMut.Lock()
	for _, c := range s.conns {
		if c.version != s.data.version {
			go c.push()
		}
	}
	s.connMut.Unlock()
	//defered cleanup()
}
func ( s * State ) gopush ( ) { s . push . mut . Lock ( ) t0 := time . Now ( ) //queue cleanup defer func ( ) { //measure time passed, ensure we wait at least Throttle time tdelta := time . Now ( ) . Sub ( t0 ) if t := s . Throttle - tdelta ; t > 0 { time . Sleep ( t ) } //push complete s . push . mut . Unlock ( ) atomic . StoreUint32 ( & s . push . ing , 0 ) //if queued, auto-push again if atomic . CompareAndSwapUint32 ( & s . push . queued , 1 , 0 ) { s . Push ( ) } } ( ) //calculate new json state l , hasLock := s . gostruct . ( sync . Locker ) if hasLock { l . Lock ( ) } newBytes , err := json . Marshal ( s . gostruct ) if hasLock { l . Unlock ( ) } if err != nil { log . Printf ( " " , err ) return } //if changed, then calculate change set if ! bytes . Equal ( s . data . bytes , newBytes ) { //calculate change set from last version ops , _ := jsonpatch . CreatePatch ( s . data . bytes , newBytes ) if len ( s . data . bytes ) > 0 && len ( ops ) > 0 { //changes! bump version s . data . mut . Lock ( ) s . data . delta , _ = json . Marshal ( ops ) s . data . bytes = newBytes s . data . version ++ s . data . mut . Unlock ( ) } } //send this new change to each subscriber s . connMut . Lock ( ) for _ , c := range s . conns { if c . version != s . data . version { go c . push ( ) } } s . connMut . Unlock ( ) //defered cleanup() }
//non-blocking push
non - blocking push