Stubbed out primitive working copy of Disruptor.
diff --git a/barrier.go b/barrier.go
new file mode 100644
index 0000000..03d0bcf
--- /dev/null
+++ b/barrier.go
@@ -0,0 +1,5 @@
+package disruptor
+
+type Barrier interface {
+ Read(int64) int64
+}
diff --git a/cursor.go b/cursor.go
new file mode 100644
index 0000000..ec1b18c
--- /dev/null
+++ b/cursor.go
@@ -0,0 +1,25 @@
+package disruptor
+
+const (
+ InitialSequenceValue int64 = -1
+ cpuCacheLinePadding = 7
+)
+
+const (
+ Gating = -2
+ Idling = -3
+ Stopped = -4
+)
+
+type Cursor struct {
+ sequence int64
+ padding [cpuCacheLinePadding]int64
+}
+
+func NewCursor() *Cursor {
+ return &Cursor{sequence: InitialSequenceValue}
+}
+
+func (this *Cursor) Read(minimum int64) int64 {
+ return this.sequence
+}
diff --git a/reader.go b/reader.go
new file mode 100644
index 0000000..8c66037
--- /dev/null
+++ b/reader.go
@@ -0,0 +1,37 @@
+package disruptor
+
+type Reader struct {
+ read *Cursor
+ written *Cursor
+ upstream Barrier
+ ready bool
+} // TODO: padding???
+
+func NewReader(read, written *Cursor, upstream Barrier) *Reader {
+ return &Reader{
+ read: read,
+ written: written,
+ upstream: upstream,
+ ready: true,
+ }
+}
+
+func (this *Reader) Start() {
+ this.ready = true
+}
+func (this *Reader) Stop() {
+ this.ready = false
+}
+func (this *Reader) Receive(next int64) int64 {
+ maximum := this.upstream.Read(next)
+
+ if next <= maximum {
+ return maximum
+ } else if maximum = this.written.sequence; next <= maximum {
+ return Gating
+ } else if this.ready {
+ return Idling
+ } else {
+ return Stopped
+ }
+}
diff --git a/writer.go b/writer.go
new file mode 100644
index 0000000..7636b07
--- /dev/null
+++ b/writer.go
@@ -0,0 +1,50 @@
+package disruptor
+
+type Writer struct {
+ written *Cursor
+ upstream Barrier
+ capacity int64
+ previous int64
+ gate int64
+} // TODO: padding?
+
+func NewWriter(written *Cursor, upstream Barrier, capacity int64) *Writer {
+ assertPowerOfTwo(capacity)
+
+ return &Writer{
+ upstream: upstream,
+ written: written,
+ capacity: capacity,
+ previous: InitialSequenceValue,
+ gate: InitialSequenceValue,
+ }
+}
+
+func assertPowerOfTwo(value int64) {
+ if value > 0 && (value&(value-1)) != 0 {
+ // Wikipedia entry: http://bit.ly/1krhaSB
+ panic("The ring capacity must be a power of two, e.g. 2, 4, 8, 16, 32, 64, etc.")
+ }
+}
+
+func (this *Writer) Reserve() int64 {
+ next := this.previous + 1
+ wrap := next - this.capacity
+
+ if wrap > this.gate {
+ // this.gate = this.upstream.WaitFor(next) // investigate...
+ min := this.upstream.Read(next)
+ if wrap > min {
+ return Gating
+ }
+
+ this.gate = min
+ }
+
+ this.previous = next
+ return next
+}
+
+func (this *Writer) Commit(sequence int64) {
+ this.written.sequence = sequence
+}
