<commit_before>/*
   Copyright 2014 CoreOS, Inc.

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package rafthttp

import (
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"net/url"
	"path"
	"strconv"
	"sync"
	"time"

	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/raft/raftpb"

	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
)

const (
	streamBufSize = 4096
)

// TODO: a stream might hava one stream server or one stream client, but not both.
type stream struct {
	// the server might be attached asynchronously with the owner of the stream
	// use a mutex to protect it
	sync.Mutex
	w *streamWriter

	r *streamReader
}

func (s *stream) open(from, to, cid types.ID, term uint64, tr http.RoundTripper, u string, r Raft) error {
	if s.r != nil {
		panic("open: stream is open")
	}

	c, err := newStreamReader(from, to, cid, term, tr, u, r)
	if err != nil {
		log.Printf("stream: error opening stream: %v", err)
		return err
	}
	s.r = c
	return nil
}

func (s *stream) attach(sw *streamWriter) error {
	s.Lock()
	defer s.Unlock()
	if s.w != nil {
		// ignore lower-term streaming request
		if sw.term < s.w.term {
			return fmt.Errorf("cannot attach out of data stream server [%d / %d]", sw.term, s.w.term)
		}
		s.w.stop()
	}
	s.w = sw
	return nil
}

func (s *stream) write(m raftpb.Message) bool {
	s.Lock()
	defer s.Unlock()
	if s.w == nil {
		return false
	}
	if m.Term != s.w.term {
		if m.Term > s.w.term {
			panic("expected server to be invalidated when there is a higher term message")
		}
		return false
	}
	// todo: early unlock?
	if err := s.w.send(m.Entries); err != nil {
		log.Printf("stream: error sending message: %v", err)
		log.Printf("stream: stopping the stream server...")
		s.w.stop()
		s.w = nil
		return false
	}
	return true
}

// invalidate stops the sever/client that is running at
// a term lower than the given term.
func (s *stream) invalidate(term uint64) {
	s.Lock()
	defer s.Unlock()

	if s.w != nil {
		if s.w.term < term {
			s.w.stop()
			s.w = nil
		}
	}
	if s.r != nil {
		if s.r.term < term {
			s.r.stop()
			s.r = nil
		}
	}
}

func (s *stream) stop() {
	s.invalidate(math.MaxUint64)
}

func (s *stream) isOpen() bool {
	if s.r != nil && s.r.isStopped() {
		s.r = nil
	}
	return s.r != nil
}

type WriteFlusher interface {
	io.Writer
	http.Flusher
}

// TODO: replace fs with stream stats
type streamWriter struct {
	to   types.ID
	term uint64
	fs   *stats.FollowerStats<commit_msg>rafthttp: a stopped stream does not accept any methods<commit_after>/*
   Copyright 2014 CoreOS, Inc.

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package rafthttp

import (
	"errors"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"net/url"
	"path"
	"strconv"
	"sync"
	"time"

	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/raft/raftpb"

	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
)

const (
	streamBufSize = 4096
)

// TODO: a stream might hava one stream server or one stream client, but not both.
type stream struct {
	sync.Mutex
	w       *streamWriter
	r       *streamReader
	stopped bool
}

func (s *stream) open(from, to, cid types.ID, term uint64, tr http.RoundTripper, u string, r Raft) error {
	c, err := newStreamReader(from, to, cid, term, tr, u, r)
	if err != nil {
		log.Printf("stream: error opening stream: %v", err)
		return err
	}

	s.Lock()
	defer s.Unlock()
	if s.stopped {
		return errors.New("stream: stopped")
	}
	if s.r != nil {
		panic("open: stream is open")
	}
	s.r = c
	return nil
}

func (s *stream) attach(sw *streamWriter) error {
	s.Lock()
	defer s.Unlock()
	if s.stopped {
		return errors.New("stream: stopped")
	}
	if s.w != nil {
		// ignore lower-term streaming request
		if sw.term < s.w.term {
			return fmt.Errorf("cannot attach out of data stream server [%d / %d]", sw.term, s.w.term)
		}
		s.w.stop()
	}
	s.w = sw
	return nil
}

func (s *stream) write(m raftpb.Message) bool {
	s.Lock()
	defer s.Unlock()
	if s.stopped {
		return false
	}
	if s.w == nil {
		return false
	}
	if m.Term != s.w.term {
		if m.Term > s.w.term {
			panic("expected server to be invalidated when there is a higher term message")
		}
		return false
	}
	// todo: early unlock?
	if err := s.w.send(m.Entries); err != nil {
		log.Printf("stream: error sending message: %v", err)
		log.Printf("stream: stopping the stream server...")
		s.w.stop()
		s.w = nil
		return false
	}
	return true
}

// invalidate stops the sever/client that is running at
// a term lower than the given term.
func (s *stream) invalidate(term uint64) {
	s.Lock()
	defer s.Unlock()
	if s.w != nil {
		if s.w.term < term {
			s.w.stop()
			s.w = nil
		}
	}
	if s.r != nil {
		if s.r.term < term {
			s.r.stop()
			s.r = nil
		}
	}
	if term == math.MaxUint64 {
		s.stopped = true
	}
}

func (s *stream) stop() {
	s.invalidate(math.MaxUint64)
}

func (s *stream) isOpen() bool {
	s.Lock()
	defer s.Unlock()
	if s.r != nil && s.r.isStopped() {
		s.r = nil
	}
	return s.r != nil
}

type WriteFlusher interface {
	io.Writer
	http.Flusher
}

// TODO: replace fs with stream stats
type streamWriter struct {
	to   types.ID
	term uint64
	fs   *stats.FollowerStats
{'fraction_non_alphanumeric': 0.12802435507130266, 'fraction_numerical': 0.006729690754686749, 'mean_word_length': 2.9808673469387754, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1835524', 'n_tokens_mistral': 2413, 'n_tokens_neox': 2301, 'n_words': 751}