av/ringbuffer/RingBuffer.go

/*
NAME
  RingBuffer.go - a structure that encapsulates a RingBuffer datastructure with conccurency
  functionality

DESCRIPTION
  See Readme.md

AUTHOR
  Saxon Nelson-Milton <saxon.milton@gmail.com>

LICENSE
  RingBuffer.go is Copyright (C) 2017 the Australian Ocean Lab (AusOcean)

  It is free software: you can redistribute it and/or modify them
  under the terms of the GNU General Public License as published by the
  Free Software Foundation, either version 3 of the License, or (at your
  option) any later version.

  It is distributed in the hope that it will be useful, but WITHOUT
  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
 for more details.

  You should have received a copy of the GNU General Public License
  along with revid in gpl.txt.  If not, see [GNU licenses](http://www.gnu.org/licenses).
*/

package ringbuffer

import (
	"errors"
	"sync"
)

/*
RingBuffer aims to provide functionality of a RingBuffer data structure.
It may be used in concurrent routines.
*/
type ringBuffer struct {
	dataMemory       [][]byte
	sizeMemory       []int
	size             int
	noOfElements     int
	first            int
	last             int
	currentlyWriting bool
	currentlyReading bool
	sizeChannel      chan int
	mutex            sync.Mutex
}

/*
New returns the addres of a new ring buffer with the parameters specified.
It initialises fields and allocates the required dataMemory.
*/
func NewRingBuffer(size int, elementSize int) (rb *ringBuffer) {
	if size <= 0 || elementSize <= 0 {
		return nil
	}
	rb = new(ringBuffer)
	rb.dataMemory = make([][]byte, size)
	for i := range rb.dataMemory {
		rb.dataMemory[i] = make([]byte, elementSize)
	}
	rb.sizeMemory = make([]int, size)
	rb.size = size
	rb.noOfElements = 0
	rb.first = -1
	rb.last = -1
	rb.currentlyWriting = false
	rb.currentlyReading = false
	rb.sizeChannel = make(chan int, size)
	rb.mutex = sync.Mutex{}
	return
}

/*
Get provides the address to the next empty element in the RingBuffer.
An error is returned if the buffer is full, or there has already been
a call to Get without an accompanying call to DoneWriting.
*/
func (rb *ringBuffer) Get() ([]byte, error) {
	rb.mutex.Lock()
	defer rb.mutex.Unlock()
	if rb.noOfElements == rb.size {
		return nil, errors.New("Buffer full!")
	}
	if rb.currentlyWriting {
		return nil, errors.New("Second call to Get! Call DoneWriting first!")
	}
	rb.currentlyWriting = true
	nextlast := rb.last + 1
	if nextlast == rb.size {
		nextlast = 0
	}
	return rb.dataMemory[nextlast], nil
}

/*
DoneWriting let's the RingBuffer know that we have finished writing to the
address recieved by a call to Get, and also let's the buffer know how much
data was written. An Error is thrown if there was not an initial call to Get.
*/
func (rb *ringBuffer) DoneWriting(size int) error {
	rb.mutex.Lock()
	defer rb.mutex.Unlock()
	if !rb.currentlyWriting {
		return errors.New("DoneWriting called without initial call to Get!")
	}
	if rb.first == -1 {
		rb.first++
	}
	rb.last++
	if rb.last == rb.size {
		rb.last = 0
	}
	rb.sizeMemory[rb.last] = size 	 	
	rb.noOfElements++
	rb.currentlyWriting = false
	return nil
}

/*
Read gets the address of the next element that we would like to get data
from and thereafter consider empty in the buffer for writing during latter
calls to Get. The address of the data is returned, as well as the size of
the data contained at this address. An error is returned if the buffer is
empty, or there has been a second call to Read before a call to DoneReading.
*/
func (rb *ringBuffer) Read() ([]byte, error) {
	rb.mutex.Lock()
	defer rb.mutex.Unlock()
	if !rb.CanRead() {
		return nil, errors.New("Buffer is empty, nothging to read!")
	}
	if rb.currentlyReading {
		return nil, errors.New("Second call to Read! Call DoneReading first!")
	}
	rb.currentlyReading = true
	return rb.dataMemory[rb.first][:rb.sizeMemory[rb.first]], nil
}

/*
DoneReading informs the buffer that we have finished with the address
provided by Read. The buffer can now consider this address to be empty.
An error is returned if there was no initial call to Read.
*/
func (rb *ringBuffer) DoneReading() error {
	rb.mutex.Lock()
	defer rb.mutex.Unlock()
	if !rb.currentlyReading {
		return errors.New("DoneReading called but no initial call to Read!")
	}
	rb.first++
	if rb.first == rb.size {
		rb.first = 0
	}
	rb.noOfElements--
	rb.currentlyReading = false
	return nil
}

/*
CanRead returns true if it is possible to read from the buffer, i.e. if
it is not empty.
*/
func (rb *ringBuffer) CanRead() bool {
	if rb.first == -1 || rb.noOfElements == 0 {
		return false
	}
	return true
}