Control.Pipeline

Pipelining is sending multiple requests over a socket and receiving the responses later, in the same order. This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a promise (future) response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipeline (and get promises back); it will pipeline each thread's request right away without waiting.

A pipeline closes itself when a read or write causes an error, so you can detect a broken pipeline by checking isClosed. It also closes itself when garbage collected, or you can close it explicitly.

Pipeline

data Pipeline handle bytes

Thread-safe and pipelined socket

Instances

Resource IO h => Resource IO (Pipeline h b)

newPipeline

Arguments

:: (Stream h b, Resource IO h)

=> (Size -> b)

Convert Size to bytes of fixed length. Every Int must translate to same number of bytes.

-> (b -> Size)

Convert bytes of fixed length to Size. Must be exact inverse of encodeSize.

-> h

Underlying socket (handle) this pipeline will read/write from

-> IO (Pipeline h b)

Create new Pipeline with given encodeInt, decodeInt, and handle. You should close pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.

send :: (Stream h b, Resource IO h) => Pipeline h b -> [b] -> IO ()

Send messages all together to destination (no messages will be interleaved between them). None of the messages can induce a response, i.e. the destination must not reply to any of these messages (otherwise future calls will get these responses instead of their own). Each message is preceeded by its length when written to socket. Raises IOError and closes pipeline if send fails

call :: (Stream h b, Resource IO h) => Pipeline h b -> [b] -> IO (IO b)

Send messages all together to destination (no messages will be interleaved between them), and return promise of response from one message only. One and only one message in the list must induce a response, i.e. the destination must reply to exactly one message only (otherwise promises will have the wrong responses in them). Each message is preceeded by its length when written to socket. Likewise, the response must be preceeded by its length. Raises IOError and closes pipeline if send fails, likewise for reply.

Util

type Size = Int

class Length list where

Methods

length :: list -> Size

Instances

Length ByteString

Length ByteString

class Resource m r where

Methods

close :: r -> m ()

Close resource

isClosed :: r -> m Bool

Is resource closed

Instances

Resource IO Handle

Access m => Resource m Cursor

Resource IO h => Resource IO (Pipeline h b)

class Flush handle where

Methods

flush :: handle -> IO ()

Flush written bytes to destination

Instances

Flush Handle

class (Length bytes, Monoid bytes, Flush handle) => Stream handle bytes where

Methods

put :: handle -> bytes -> IO ()

Write bytes to handle

get :: handle -> Int -> IO bytes

Read up to N bytes from handle; if EOF return empty bytes, otherwise block until at least 1 byte is available

Instances

Stream Handle ByteString

Stream Handle ByteString

getN :: Stream h b => h -> Int -> IO b

Read N bytes from hande, blocking until all N bytes are read. If EOF is reached before N bytes then throw EOF exception.