'From Cuis 1.0 of 6 August 2009 [latest update: #258] on 20 August 2009 at 10:28:12 pm'!

!Socket commentStamp: 'gk 12/13/2005 00:43' prior: 0!
A Socket represents a network connection point. Current sockets are designed to support the TCP/IP and UDP protocols. Sockets are the lowest level of networking object in Squeak and are not normally used directly. SocketStream is a higher level object wrapping a Socket in a stream like protocol.

ProtocolClient and subclasses are in turn wrappers around a SocketStream to provide support for specific network protocols such as POP, NNTP, HTTP, and FTP.!

!classDefinition: #SocketStream category: #'Network-Kernel'!
Object subclass: #SocketStream
	instanceVariableNames: 'socket inStream outStream timeout autoFlush buffered bufferSize outNextToWrite binary inNextToWrite lastRead recentlyRead outBuffer inBuffer shouldSignal '
	classVariableNames: ''
	poolDictionaries: ''
	category: 'Network-Kernel'!

!SocketStream commentStamp: 'md 7/14/2006 16:32' prior: 0!
SocketStream is a wrapper for class Socket making it easy to write networking code by giving the programmer a stream-like protocol. A Socket is a two way communication link with two logically separate channels - input and output. The Socket class is the lowest level in Squeak for network communication and using it directly can be difficult and bug prone.

A SocketStream can be in binary or ascii mode, ascii is the default which means you are transmitting and receiving Strings. Most Internet protocols are in clear text ascii, like for example HTTP. Another setting is what timeout you want to use - default is the standardTimeout from Socket. More settings can be found in the method category 'configuration'.

Simplest example of connecting, sending/receiving and closing:

| stream result |
stream := SocketStream openConnectionToHostNamed: 'www.squeak.org' port: 80.
[[stream nextPutAll: 'GET / HTTP/1.0'; crlf; crlf; flush.
result := stream upToEnd. "Give us all data until the socket is closed."
Transcript show: result; cr.]
	ensure: [stream close]]
		on: ConnectionTimedOut
		do: [:ex | Transcript show: ex asString;cr. ex resume]

There are two important things to note above:
	- The methods in category "stream in" can signal two exceptions (unless turned off with #shouldSignal:):
		ConnectionClosed and ConnectionTimedOut
	- We close the stream using #ensure:, that is to make sure it isn't left opened.
	- We use #on:do: to catch any signal. In this case we do not need to catch ConnectionClosed since #upToEnd does that for us intrinsically.

----------------
SocketStream (below called SS) is a reimplementation of 'Old'-SocketStream (below called OSS) - the class that originates from the original Comanche implementation but now is included in standard Squeak. SS has the same protocol as OSS and is meant to replace it. SS is faster, more flexible, is better documented and adds a few features:

1. #shouldSignal:, which decides if SS should signal low level Socket exceptions (true) or if it should swallow them like original OSS did. Default is true. The only reason I added this is for backwards compatibility - not signalling causes problems - see bug 4 below.

2. #nextAllInBuffer, #nextInBuffer:, #skip:, #receiveData:, #nextPutAllFlush: and #recentlyRead are new additions to the public protocol.


It also fixes various bugs:

1. #isDataAvailable could theoretically answer false, when there actually is some in the buffer in OSS. If #receiveDataIfAvailable reads the last byte then the following "socket dataAvailable" would answer false. So the last byte would be sitting in the inStream missed.

2. #upToAll: in OSS has several problems, for example - #positionOfSubCollection:ifAbsent: which was introduced answers one position too low. This was compensated in upToAll:, but only in the pushBack: call, not the actual result being returned which was cut short 1 byte. Amusingly this makes KomHttpServer not use "Keep-Alive" since the last $e in 'Alive' was cut short. :)

3. SS doesn't inherit from PositionableStream since that just breaks various inherited messages, like for example #skip:. OSS should IMHO be changed to inherit from Object - or of course, replaced in full with SS. :)

4. Since SocketStream by default signals closes and timeouts the SocketStreamTest now passes. The reason for SocketStream to fail is that while it does timeout on a low level (#SocketStream>>receiveData doesn't hang forever) - the callers of #receiveData sometimes loop - like in #next:, and thus eliminates the timeout. SS warns about some methods (in their method comments) not honouring timeouts if shouldSignal is false, I really don't know what they should do in that case:
	#next:, #upTo:, #upToAll: and #upToEnd (and #receiveData:)


The primary reason for the SS implementation is optimal performance. The main differences in implementation with the old OSS are:

1. SS uses two buffers directly (inBuffer and outBuffer) with pointers marking start and stop within the buffer. OSS instead uses two regular streams, a ReadStream and a WriteStream. Using internal buffers makes it possible to avoid copying and reallocation in various ways, it also makes SS be able to have specialized growing/buffer moving behaviour.

2. #upTo:, #upToAll: and #peekForAll: uses selectged String messages that in turn uses fast primitives for searching. OSS used other messages that fell back on byte per byte reading.

3. #receiveData in OSS creates a temporary buffer stream for each call!! During a long read operation, like say #upToAll: (which for example is used when uploading files using HTTP POST forms), this is devastating - especially since the default size is only 2000 bytes - and leads to a very high number of low level read operations on the Socket, typically 100 times more calls than with OSS. The buffer in OSS is held in an instvar (not recreated for each call), is larger from the start and above all - grows dynamically by doubling. OSS can also avoid a grow/reallocation by doing a "move down" if data has been read from the SS as it comes in and through that making room in the lower part of the inBuffer. The net result is that upToAll: for large files is about 10 times faster.

4. The implementation of upTo: and upToAll: tries to avoid doing unnecessary find operations in the buffer and is greedy by default, which means it favors reading more data - if available - before searching for the stop sequence. If we had #findString:startingAt:stoppingAt: this wouldn't have to be greedy and we wouldn't be needlessly scanning dead buffer area. VM hackers? Also, while you are at it - make it work for ByteArrays too. :)


SS can not be run unbuffered, since that seems unneeded. The option to autoFlush is still available, with it set to true SocketStream (just like OSS) will flush on its own on each nextPut:/nextPutAll:, otherwise flushing it will have to be done manually but is done on close.

The first performance tests shows that, as noted above, receiving large amounts of data using #upToAll: is greatly improved - factor of 10. Serving HTTP with small payloads seemed at first not be faster at all - but this is due to the high overhead of Socket connect/close and other things. Increasing payloads show a difference and especially with keep alive on - where the new SS roughly doubles the throughput!!!


!Socket methodsFor: 'connection open/close' stamp: 'mir 5/9/2003 18:13'!
connectNonBlockingTo: hostAddress port: port
	"Initiate a connection to the given port at the given host address. This operation will return immediately; follow it with waitForConnectionUntil: to wait until the connection is established."

	| status |
	self initializeNetwork.
	status := self primSocketConnectionStatus: socketHandle.
	(status == Unconnected)
		ifFalse: [InvalidSocketStatusException signal: 'Socket status must Unconnected before opening a new connection'].

	self primSocket: socketHandle connectTo: hostAddress port: port.
! !

!Socket methodsFor: 'connection open/close' stamp: 'mir 2/22/2002 16:25'!
listenOn: port
	"Listen for a connection on the given port. This operation will return immediately; follow it with waitForConnectionUntil: to wait until a connection is established."

	| status |
	status := self primSocketConnectionStatus: socketHandle.
	(status == Unconnected)
		ifFalse: [InvalidSocketStatusException signal: 'Socket status must Unconnected before listening for a new connection'].

	self primSocket: socketHandle listenOn: port.
! !

!Socket methodsFor: 'connection open/close' stamp: 'mir 2/22/2002 16:25'!
listenOn: portNumber backlogSize: backlog
	"Listen for a connection on the given port.
	If this method succeeds, #accept may be used to establish a new connection"
	| status |
	status := self primSocketConnectionStatus: socketHandle.
	(status == Unconnected)
		ifFalse: [InvalidSocketStatusException signal: 'Socket status must Unconnected before listening for a new connection'].
	self primSocket: socketHandle listenOn: portNumber backlogSize: backlog.
! !

!Socket methodsFor: 'connection open/close' stamp: 'ikp 9/1/2003 20:32'!
listenOn: portNumber backlogSize: backlog interface: ifAddr
	"Listen for a connection on the given port.
	If this method succeeds, #accept may be used to establish a new connection"
	| status |
	status := self primSocketConnectionStatus: socketHandle.
	(status == Unconnected)
		ifFalse: [InvalidSocketStatusException signal: 'Socket status must Unconnected before listening for a new connection'].
	self primSocket: socketHandle listenOn: portNumber backlogSize: backlog interface: ifAddr.
! !

!Socket methodsFor: 'receiving' stamp: 'gk 12/14/2005 10:02'!
discardReceivedData
	"Discard any data received up until now, and return the number of bytes discarded."

	| buf totalBytesDiscarded |
	buf := String new: 10000.
	totalBytesDiscarded := 0.
	[self isConnected] whileTrue: [
		totalBytesDiscarded :=
			totalBytesDiscarded + (self receiveDataInto: buf)].
	^ totalBytesDiscarded
! !

!Socket methodsFor: 'receiving' stamp: 'mu 8/9/2003 18:04'!
receiveAvailableDataInto: buffer startingAt: startIndex
	"Receive all available data into the given buffer and return the number of bytes received.
	Note the given buffer may be only partially filled by the received data.
	Do not wait for data."

	| bufferPos bytesRead |
	bufferPos := startIndex.
	[self dataAvailable
		and: [bufferPos-1 < buffer size]] 
		whileTrue: [
			bytesRead := self receiveSomeDataInto: buffer startingAt: bufferPos.
			bufferPos := bufferPos + bytesRead].
	^bufferPos - startIndex! !

!Socket methodsFor: 'receiving' stamp: 'yo 10/10/2005 18:47'!
receiveAvailableDataIntoBuffer: buffer
	"Receive all available data (if any). Do not wait."
 
	| bytesRead |
	bytesRead := self receiveAvailableDataInto: buffer.
	^buffer copyFrom: 1 to: bytesRead! !

!Socket methodsFor: 'receiving' stamp: 'svp 9/23/2003 00:12'!
receiveDataInto: aStringOrByteArray startingAt: aNumber
	"Receive data into the given buffer and return the number of bytes received. 
	Note the given buffer may be only partially filled by the received data.
	Waits for data once.  The answer may be zero (indicating that no data was 
	available before the socket closed)."

	| bytesRead closed |
	bytesRead := 0.
	closed := false.
	[closed not and: [bytesRead == 0]]
		whileTrue: [
			self waitForDataIfClosed: [closed := true].
			bytesRead := self primSocket: socketHandle
				receiveDataInto: aStringOrByteArray
				startingAt: aNumber
				count: aStringOrByteArray size-aNumber+1].
	^bytesRead
! !

!Socket methodsFor: 'receiving' stamp: 'gk 2/9/2005 12:33'!
receiveDataSignallingClosedInto: aStringOrByteArray startingAt: aNumber
	"Receive data into the given buffer and return the number of bytes received. 
	Note the given buffer may be only partially filled by the received data.
	Waits for data until something is read or the socket is closed, upon which
	we signal."

	| bytesRead |
	bytesRead := 0.
	[bytesRead == 0]
		whileTrue: [
			self waitForData.
			bytesRead := self primSocket: socketHandle
				receiveDataInto: aStringOrByteArray
				startingAt: aNumber
				count: aStringOrByteArray size-aNumber+1].
	^bytesRead
! !

!Socket methodsFor: 'receiving' stamp: 'gk 2/9/2005 12:24'!
receiveDataSignallingTimeout: timeout into: aStringOrByteArray startingAt: aNumber
	"Receive data into the given buffer and return the number of bytes received. 
	Note the given buffer may be only partially filled by the received data.
	Wait for data once for the specified nr of seconds.  This method will
	throw exceptions on timeout or the socket closing."

	self waitForDataFor: timeout.
	^self primSocket: socketHandle
		receiveDataInto: aStringOrByteArray
		startingAt: aNumber
		count: aStringOrByteArray size-aNumber+1
! !

!Socket methodsFor: 'waiting' stamp: 'mu 8/9/2003 15:17'!
waitForAcceptFor: timeout
	"Wait and accept an incoming connection. Return nil if it falis"
	[self waitForConnectionFor: timeout] on: ConnectionTimedOut do: [:ex | ^nil].
	^self isConnected
		ifTrue:[self accept]
		! !

!Socket methodsFor: 'waiting' stamp: 'gk 12/14/2005 09:59'!
waitForDisconnectionFor: timeout
	"Wait for the given nr of seconds for the connection to be broken.
	Return true if it is broken by the deadline, false if not.
	The client should know the connection is really going to be closed
	(e.g., because he has called 'close' to send a close request to the other end)
	before calling this method."

	| status deadline |
	status := self primSocketConnectionStatus: socketHandle.
	deadline := Socket deadlineSecs: timeout.
	[((status == Connected) or: [(status == ThisEndClosed)]) and:
	 [Time millisecondClockValue < deadline]] whileTrue: [
		self discardReceivedData.
		self readSemaphore waitTimeoutMSecs: (deadline - Time millisecondClockValue).
		status := self primSocketConnectionStatus: socketHandle].

	^ status ~= Connected
! !


!Socket class methodsFor: 'tests' stamp: 'gk 12/15/2005 01:18'!
loopbackTest
	"Send data from one socket to another on the local machine.
	Tests most of the socket primitives."

	"100 timesRepeat: [Socket loopbackTest]"

	| sock1 sock2 bytesToSend sendBuf receiveBuf done bytesSent bytesReceived t extraBytes packetsSent packetsRead |
	Transcript
		cr;
		show: 'starting loopback test';
		cr.
	Transcript
		show: '---------- Connecting ----------';
		cr.
	self initializeNetwork.
	sock1 := self new.
	sock2 := self new.
	sock1 listenOn: 54321.
	sock2 connectTo: NetNameResolver localHostAddress port: 54321.
	sock1 waitForConnectionFor: self standardTimeout.
	sock2 waitForConnectionFor: self standardTimeout.
	sock1 isConnected ifFalse: [self error: 'sock1 not connected'].
	sock2 isConnected ifFalse: [self error: 'sock2 not connected'].
	Transcript
		show: 'connection established';
		cr.
	bytesToSend := 5000000.
	sendBuf := String new: 5000 withAll: $x.
	receiveBuf := String new: 50000.
	done := false.
	packetsSent := packetsRead := bytesSent := bytesReceived := 0.
	t := Time millisecondsToRun: 
					[[done] whileFalse: 
							[(sock1 sendDone and: [bytesSent < bytesToSend]) 
								ifTrue: 
									[packetsSent := packetsSent + 1.
									bytesSent := bytesSent + (sock1 sendSomeData: sendBuf)].
							sock2 dataAvailable 
								ifTrue: 
									[packetsRead := packetsRead + 1.
									bytesReceived := bytesReceived + (sock2 receiveDataInto: receiveBuf)].
							done := bytesSent >= bytesToSend and: [bytesReceived = bytesSent]]].
	Transcript
		show: 'closing connection';
		cr.
	sock1 waitForSendDoneFor: self standardTimeout.
	sock1 close.
	sock2 waitForDisconnectionFor: self standardTimeout.
	extraBytes := sock2 discardReceivedData.
	extraBytes > 0 
		ifTrue: 
			[Transcript
				show: ' *** received ' , extraBytes size printString , ' extra bytes ***';
				cr].
	sock2 close.
	sock1 waitForDisconnectionFor: self standardTimeout.
	sock1 isUnconnectedOrInvalid ifFalse: [self error: 'sock1 not closed'].
	sock2 isUnconnectedOrInvalid ifFalse: [self error: 'sock2 not closed'].
	Transcript
		show: '---------- Connection Closed ----------';
		cr.
	sock1 destroy.
	sock2 destroy.
	Transcript
		show: 'loopback test done; time = ' , t printString;
		cr.
	Transcript
		show: (bytesToSend asFloat / t roundTo: 0.01) printString 
					, '* 1000 bytes/sec';
		cr.
	Transcript endEntry! !

!Socket class methodsFor: 'tests' stamp: 'gk 12/15/2005 01:03'!
newAcceptCheck
	"Check if the platform has support for the BSD style accept()."

	"Socket newAcceptCheck"
	
	| socket |
	self initializeNetwork.
	socket := self newTCP.
	socket listenOn: 44444 backlogSize: 4.
	socket isValid ifTrue: [
		self inform: 'Everything looks OK for the BSD style accept()'
	] ifFalse: [
		self inform: 'It appears that you DO NOT have support for the BSD style accept()'].
	socket destroy! !

!Socket class methodsFor: 'tests' stamp: 'jmv 8/18/2009 10:14'!
sendTest
	"Send data to the 'discard' socket of the given host.
	Tests the speed of one-way data transfers across the
	network to the given host. Note that most hosts
	do not run a discard server."

	"Socket sendTest"

	| sock bytesToSend sendBuf bytesSent t serverName serverAddr |
	Transcript cr; show: 'starting send test'; cr.
	self initializeNetwork.
	serverName := FillInTheBlank request: 'What is the destination server?' initialAnswer: 'create.ucsb.edu'.
	serverAddr := NetNameResolver addressForName: serverName timeout: 10.
	serverAddr = nil 
		ifTrue: [^self inform: 'Could not find an address for ' , serverName].
	sock := self new.
	Transcript show: '---------- Connecting ----------';cr.
	sock connectTo: serverAddr port: 9.
	sock isConnected ifFalse: [
		sock destroy.
		^self inform: 'could not connect'].
	Transcript show: 'connection established; sending data'; cr.
	bytesToSend := 1000000.
	sendBuf := String new: 64 * 1024 withAll: $x.
	bytesSent := 0.
	t := Time millisecondsToRun: 
					[[bytesSent < bytesToSend] whileTrue: 
							[sock sendDone 
								ifTrue: [bytesSent := bytesSent + (sock sendSomeData: sendBuf)]]].
	sock waitForSendDoneFor: self standardTimeout.
	sock destroy.
	Transcript show: '---------- Connection Closed ----------'; cr;
		show: 'send test done; time = ' , t printString; cr;
		show: (bytesToSend asFloat / t roundTo: 0.01) printString, ' * 1000 bytes/sec';cr;endEntry! !


!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 13:33'!
next
	"Return next byte, if inBuffer is empty
	we recieve some more data and try again."

	self atEnd ifTrue: [^nil].
	self isInBufferEmpty ifTrue:
		[self receiveData.
		self atEnd ifTrue: [^nil]].
	lastRead := lastRead + 1.
	^inBuffer at: lastRead! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:17'!
next: anInteger
	"Answer anInteger bytes of data.

	NOTE: This method doesn't honor timeouts if shouldSignal is false!!"

	| start |
	self receiveData: anInteger.
	start := lastRead + 1.
	lastRead := (lastRead + anInteger) min: inNextToWrite - 1.
	^inBuffer copyFrom: start to: lastRead! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 21:26'!
nextAllInBuffer
	"Return all data currently in the inBuffer,"

	^self nextInBuffer: inNextToWrite - lastRead - 1! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 13:47'!
nextAvailable
	"Answer all the data currently available,
	in buffer or in socket."

	self isInBufferEmpty ifFalse: [^self nextAllInBuffer].
	self isDataAvailable ifTrue: [self receiveData].
	^self nextAllInBuffer! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 13:02'!
nextAvailable: howMany
	"Answer all the data currently available,
	in buffer or in socket - but limited to <howMany>."

	self isInBufferEmpty ifFalse: [^self nextInBuffer: howMany].
	self isDataAvailable ifTrue: [self receiveData].
	^self nextInBuffer: howMany! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 12:51'!
nextInBuffer: anInteger
	"Answer anInteger bytes of data at most,
	but only from the inBuffer."

	| start amount |
	amount := anInteger min: (inNextToWrite - lastRead - 1).
	start := lastRead + 1.
	lastRead := lastRead + amount.
	^inBuffer copyFrom: start to: lastRead! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/24/2005 12:01'!
nextLineCrLf
	^self upToAll: String crlf! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:09'!
peek
	"Return next byte, if inBuffer is empty
	we recieve some more data and try again.
	Do not consume the byte."

	self atEnd ifTrue: [^nil].
	self isInBufferEmpty ifTrue:
		[self receiveData.
		self atEnd ifTrue: [^nil]].
	^inBuffer at: lastRead! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:16'!
peek: anInteger
	"Answer anInteger bytes of data.
	Do not consume data.

	NOTE: This method doesn't honor timeouts if shouldSignal is false!!"

	| start |
	self receiveData: anInteger.
	start := lastRead + 1.
	^inBuffer copyFrom: start to: ((lastRead + anInteger) min: inNextToWrite - 1).! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/7/2005 14:04'!
peekFor: aCharacterOrByte
	"Read and return next character or byte
	if it is equal to the argument.
	Otherwise return false."

	| nextObject |
	self atEnd ifTrue: [^false].
	self isInBufferEmpty ifTrue: 
		[self receiveData.
		self atEnd ifTrue: [^false]].
	nextObject := inBuffer at: lastRead.
	nextObject = aCharacterOrByte ifTrue: [
		lastRead := lastRead + 1.
		^true].
	^false
! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:16'!
peekForAll: aString
	"Answer whether or not the next string of characters in the receiver
	matches aString. If a match is made, advance over that string in the receiver and
	answer true. If no match, then leave the receiver alone and answer false.
	We use findString:startingAt: to avoid copying.

	NOTE: This method doesn't honor timeouts if shouldSignal is false!!"

	| sz start |
	sz := aString size.
	self receiveData: sz.
	(inNextToWrite - lastRead - 1) < sz ifTrue: [^false].
	start := lastRead + 1.
	(inBuffer findString: aString startingAt: start) = start
		ifFalse: [^false].
	lastRead := lastRead + sz.
	^true! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:15'!
skip: anInteger
	"Skip a number of bytes.
	This is faster than #next: since it does not
	have to copy and return a new String or ByteArray.

	NOTE: This method doesn't honor timeouts if shouldSignal is false!!"

	self receiveData: anInteger.
	lastRead := (lastRead + anInteger) min: inNextToWrite - 1! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 9/9/2005 02:57'!
upTo: aCharacterOrByte
	"Return data up to, but not including given character or byte.
	
	NOTE: Does not honour timeouts if shouldSignal is false!!
	
	This method looks a bit complicated, and this is mainly because there is no fast search method
	in String that takes a stoppingAt: argument. This means we need to ignore getting hits in the
	dead buffer area above inNextToWrite.
	Another measure is that this implementation is greedy and will load data into the buffer
	until there is nothing more available, or it has loaded 100kb - and not until then we search the buffer.

	A totally non greedy variant would search on every loop."

	| index result lastRecentlyRead searchedSoFar |
	searchedSoFar := 0.
	lastRecentlyRead := 0.
	index := 0.
	[self atEnd not and: [
		((lastRecentlyRead = 0 and: [self isInBufferEmpty not]) or: [self inBufferSize > 100000]) ifTrue: [
			"Data begins at lastRead + 1, we add searchedSoFar as offset."
			index := inBuffer indexOf: aCharacterOrByte startingAt: lastRead + searchedSoFar + 1.
			searchedSoFar := self inBufferSize.
			(index > 0 and: [(index + 1) > inNextToWrite]) ifTrue: [
				"Oops, hit in dead buffer area.
				This is probably due to old data, so we ignore it.
				No point in cleaning the dead area to avoid hits - it will still search it."
				index := 0]].
		index = 0]]
				whileTrue: [
					recentlyRead = 0
						ifTrue: ["blocking call for now, we don't want to poll"
							self receiveData]
						ifFalse: [
							self receiveAvailableData].
					lastRecentlyRead := recentlyRead].
	index > 0
		ifTrue: ["found it"
			result := self nextInBuffer: index - lastRead - 1.
			self skip: 1.
			^ result]
		ifFalse: ["atEnd"
			^ self nextAllInBuffer]! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 9/9/2005 02:55'!
upToAll: aStringOrByteArray
	"Answer a subcollection from the current access position to the occurrence (if any, but not
	inclusive) of aStringOrByteArray. If aCollection is not in the stream, answer the entire rest of
	the stream.
	
	NOTE: Does not honour timeouts if shouldSignal is false!!
	
	This method looks a bit complicated, and this is mainly because there is no fast search method
	in String that takes a stoppingAt: argument. This means we need to ignore getting hits in the
	dead buffer area above inNextToWrite.
	Another measure is that this implementation is greedy and will load data into the buffer
	until there is nothing more available, or it has loaded 100kb - and not until then we search the buffer.

	A totally non greedy variant would search on every loop."

	| index sz result lastRecentlyRead searchedSoFar |
	sz := aStringOrByteArray size.
	searchedSoFar := 0.
	lastRecentlyRead := 0.
	index := 0.
	[self atEnd not and: [
		((lastRecentlyRead = 0 and: [self isInBufferEmpty not]) or: [self inBufferSize > 100000]) ifTrue: [
			"Data begins at lastRead + 1, we add searchedSoFar as offset and backs up sz - 1
			so that we can catch any borderline hits."
			index := inBuffer indexOfSubCollection: aStringOrByteArray
						startingAt: lastRead + searchedSoFar - sz + 2.
			searchedSoFar := self inBufferSize.
			(index > 0 and: [(index + sz) > inNextToWrite]) ifTrue: [
				"Oops, hit partially or completely in dead buffer area.
				This is probably due to old data, so we ignore it.
				No point in cleaning the dead area to avoid hits - it will still search it."
				index := 0]].
		index = 0]]
				whileTrue: [
					recentlyRead = 0
						ifTrue: ["blocking call for now, we don't want to poll"
							self receiveData]
						ifFalse: [
							self receiveAvailableData].
					lastRecentlyRead := recentlyRead].
	index > 0
		ifTrue: ["found it"
			result := self nextInBuffer: index - lastRead - 1.
			self skip: sz.
			^ result]
		ifFalse: ["atEnd"
			^ self nextAllInBuffer]! !

!SocketStream methodsFor: 'stream in' stamp: 'gk 2/15/2005 14:14'!
upToEnd
	"Answer all data coming in on the socket until the socket
	is closed by the other end, or we get a timeout.
	This means this method catches ConnectionClosed by itself.
	
	NOTE: Does not honour timeouts if shouldSignal is false!!"

	[[self atEnd] whileFalse: [self receiveData]]
		on: ConnectionClosed
		do: [:ex | "swallow it"]. 
	^self nextAllInBuffer! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/25/2005 14:23'!
atEnd
	"There is nothing more to read when
	there is no more data in our inBuffer, the socket
	is disconnected and there is none available on the socket.
	Note that we need to check isConnected before isDataAvailable,
	otherwise data may sneak in in the meantime. But we check the
	buffer first, because it is faster."

	self isInBufferEmpty ifFalse: [^false].
	^self isConnected not
		and: [self isDataAvailable not]! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/3/2005 20:35'!
isBinary
	^binary! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/7/2005 12:24'!
isConnected
	"The stream is connected if the socket is."

	^socket isConnected! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/24/2005 12:23'!
isDataAvailable
	"It the inbuffer is empty, we check the socket for data.
	If it claims to have data available to read, we try to read
	some once and recursively call this method again.
	If something really was available it is now in the inBuffer.
	This is because there has been spurious
	dataAvailable when there really is no data to get."
 
	self isInBufferEmpty ifFalse: [^true].
	^socket dataAvailable
		ifFalse: [false]
		ifTrue: [self receiveDataIfAvailable; isDataAvailable]! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/7/2005 13:02'!
isEmpty
	"Test if there are more data to read."

	^self isInBufferEmpty and: [self isDataAvailable not]! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/7/2005 13:02'!
isInBufferEmpty
	"Any data in the buffer?"
 
	^lastRead + 1 = inNextToWrite! !

!SocketStream methodsFor: 'testing' stamp: 'gk 2/7/2005 08:59'!
isOtherEndConnected
	^socket isOtherEndClosed not! !

!SocketStream methodsFor: 'initialize-release' stamp: 'gk 2/25/2005 14:20'!
initialize
	autoFlush := true.
	shouldSignal := true.
	recentlyRead := 0.
	bufferSize := 4096.
	self ascii! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/9/2005 22:44'!
adjustInBuffer: bytesRead
	"Adjust markers and possibly grow inBuffer or move data down.
	Currently grows through doubling when less than 1024 bytes are left.
	Never shrinks. Returns the position in the buffer where any new
	data can be found."

	| old |
	bytesRead = 0 ifTrue: [^inNextToWrite].
	old := inNextToWrite.
	inNextToWrite := inNextToWrite + bytesRead.
	(inBuffer size - inNextToWrite) < 1024
		ifTrue: [
			"Hit the roof, move data down (if enough has been read) or do we grow?"
			(lastRead > 512)
				ifTrue: [^old - self moveInBufferDown]
				ifFalse: [self growInBuffer]].
	^old! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/9/2005 22:42'!
adjustOutBuffer: bytesToWrite
	"Possibly grow outBuffer to accommodate the new data.
	Currently grows through doubling when less
	than 1024 bytes are left. If bytesToWrite is even
	larger we double that instead. Never shrinks."

	(outBuffer size - outNextToWrite - bytesToWrite) < 1024 ifTrue: [
		outBuffer := (self streamBuffer: ((outBuffer size max: bytesToWrite) * 2))
						replaceFrom: 1 to: outBuffer size with: outBuffer startingAt: 1]! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/7/2005 13:09'!
checkFlush
	"If autoFlush is true we flush if
	we have reached the bufferSize
	of data in the outBuffer."

	(autoFlush and: [outNextToWrite > bufferSize])
		ifTrue: [self flush]! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/7/2005 23:05'!
growInBuffer
	"Grows through doubling."

	self resizeInBuffer: inBuffer size * 2! !

!SocketStream methodsFor: 'private' stamp: 'jmv 8/18/2009 10:30'!
moveInBufferDown
	"Move down contents of inBuffer to the start.
	Return distance moved."
	"
	jmv - Horrible hack.
	To support the nasty #pushBack: as needed by the silly XMLTokenizer, do not go to the start, but leave
	slack bytes of already read stuff.
	"

	| sz distanceMoved slack |
	
	slack _ 16.
	
	sz := inNextToWrite - lastRead - 1.
	inBuffer replaceFrom: 1+slack to: sz+slack with: inBuffer startingAt: lastRead + 1.
	distanceMoved := lastRead.
	lastRead := 0+slack.
	inNextToWrite := sz + 1+slack.
	^distanceMoved
! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/9/2005 22:36'!
resetBuffers
	"Recreate the buffers with default start sizes."

	inBuffer := self streamBuffer: bufferSize.
	lastRead := 0.
	inNextToWrite := 1.
	outBuffer := self streamBuffer: bufferSize.
	outNextToWrite := 1! !

!SocketStream methodsFor: 'private' stamp: 'gk 9/9/2005 02:29'!
resizeInBuffer: newSize
	"Resize the inBuffer by recreating it.
	This also has the effect of getting rid of
	dead data above inNextToWrite.
	<newSize> must >= inBuffer size!!"

	inBuffer := (self streamBuffer: newSize)
					replaceFrom: 1 to: inNextToWrite - 1 with: inBuffer startingAt: 1! !

!SocketStream methodsFor: 'private' stamp: 'gk 2/9/2005 22:35'!
streamBuffer: size
	"Create a buffer of the correct class and given size."

	^(self isBinary
		ifTrue: [ByteArray]
		ifFalse: [String]) new: size! !

!SocketStream methodsFor: 'stream out' stamp: 'md 2/24/2006 19:51'!
nextPut: char
	"Put a single Character or byte onto the stream."

	| toPut |
	toPut := binary ifTrue: [char asInteger] ifFalse: [char asCharacter].
	self adjustOutBuffer: 1.
	outBuffer at: outNextToWrite put: toPut.
	outNextToWrite := outNextToWrite + 1.
	self checkFlush.
	"return the argument - added by kwl"
	^ char! !

!SocketStream methodsFor: 'stream out' stamp: 'gk 2/7/2005 22:51'!
nextPutAll: aCollection
	"Put a String or a ByteArray onto the stream.
	Currently a large collection will allocate a large buffer."

	| toPut |
	toPut := binary ifTrue: [aCollection asByteArray] ifFalse: [aCollection asString].
	self adjustOutBuffer: toPut size.
	outBuffer replaceFrom: outNextToWrite to: outNextToWrite + toPut size - 1 with: toPut startingAt: 1.
	outNextToWrite := outNextToWrite + toPut size.
	self checkFlush! !

!SocketStream methodsFor: 'stream out' stamp: 'gk 2/10/2005 11:14'!
nextPutAllFlush: aCollection
	"Put a String or a ByteArray onto the stream.
	You can use this if you have very large data - it avoids
	copying into the buffer (and avoids buffer growing)
	and also flushes any other pending data first."

	| toPut |
	toPut := binary ifTrue: [aCollection asByteArray] ifFalse: [aCollection asString].
	self flush. "first flush pending stuff, then directly send"
	socket isOtherEndClosed ifFalse: [
		[socket sendData: toPut count: toPut size]
			on: ConnectionTimedOut
			do: [:ex | shouldSignal ifFalse: ["swallow"]]]! !

!SocketStream methodsFor: 'stream out' stamp: 'gk 9/9/2005 09:32'!
sendCommand: aString
	"Sends a String ending it with CR LF and then flush
	causing it to block until sent."

	self nextPutAll: aString, String crlf; flush! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/9/2005 22:37'!
ascii
	"Tell the SocketStream to send data
	as Strings instead of ByteArrays.
	This is default."

	binary := false.
	self resetBuffers! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/9/2005 22:37'!
binary
	"Tell the SocketStream to send data
	as ByteArrays instead of Strings.
	Default is ascii."

	binary := true.
	self resetBuffers! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/9/2005 22:28'!
bufferSize
	"Default buffer size is 4kb.
	increased from earlier 2000 bytes."
	
	^bufferSize! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/10/2005 17:58'!
inBufferSize
	"Answers the current size of data in the inBuffer."

	^inNextToWrite - lastRead - 1! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/10/2005 17:59'!
outBufferSize
	"Answers the current size of data in the outBuffer."

	^outNextToWrite - 1! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/10/2005 18:00'!
shouldSignal
	"If shouldSignal is enabled the Socket Exceptions
	ConnectionClosed and ConnectionTimedOut
	will not be swallowed. Default is true.
	For more info, see #shouldSignal:"

	^shouldSignal! !

!SocketStream methodsFor: 'configuration' stamp: 'gk 2/10/2005 18:03'!
shouldSignal: aBoolean
	"If shouldSignal is enabled the Socket Exceptions
	ConnectionClosed and ConnectionTimedOut will not be swallowed.
	Default is true. And please - don't set it to false - it is better to
	use an exception handler (see below)  and several methods
	in this class will not honour timeouts (says so in their method comments).
	Also, it is quite hard to understand what for example #upToEnd
	should return to indicate a timeout.
	
	Wrap your use of SocketStream with a handler like:
	
	[stuff := mySocketStream next: 10]
		on: ConnectionClosed, ConnectionTimedOut
		do: [:ex |
			Transcript show: 'Oops!! Did not get my ten bytes!!;cr]
	"

	shouldSignal := aBoolean! !

!SocketStream methodsFor: 'control' stamp: 'gk 2/24/2005 11:55'!
close
	"Flush any data still not sent
	and take care of the socket."

	self flush.
	socket closeAndDestroy: 30! !

!SocketStream methodsFor: 'control' stamp: 'gk 9/9/2005 09:33'!
flush
	"If the other end is connected and we have something
	to send, then we send it and reset the outBuffer."

	((outNextToWrite > 1) and: [socket isOtherEndClosed not])
		ifTrue: [
			[socket sendData: outBuffer count: outNextToWrite - 1]
				on: ConnectionTimedOut
				do: [:ex | shouldSignal ifFalse: ["swallow"]].
			outNextToWrite := 1]! !

!SocketStream methodsFor: 'control' stamp: 'gk 4/14/2005 09:49'!
receiveData: nBytes
	"Keep reading the socket until we have nBytes
	in the inBuffer or we reach the end. This method
	does not return data, but can be used to make sure
	data has been read into the buffer from the Socket
	before actually reading it from the FastSocketStream.
	Mainly used internally. We could also adjust the buffer
	to the expected amount of data and avoiding several
	incremental grow operations.

	NOTE: This method doesn't honor timeouts if shouldSignal
	is false!! And frankly, I am not sure how to handle that
	case or if I care - I think we should always signal."

	[self atEnd not and: [nBytes > self inBufferSize]]
		whileTrue: [self receiveData]! !

!SocketStream methodsFor: 'control' stamp: 'gk 2/9/2005 23:08'!
recentlyRead
	"Return the number of bytes read
	during the last socket operation."
	
	^recentlyRead! !

!SocketStream methodsFor: 'printing' stamp: 'gk 2/25/2005 14:19'!
debug
	"Display debug info."

	| data |
	data := self inBufferSize.
	^String streamContents: [:s |
		s
			nextPutAll: 'Buffer size: ', inBuffer size asString;cr;
			nextPutAll: 'InBuffer data size: ', data asString; cr;
			nextPutAll: 'In data (20):', (inBuffer copyFrom: lastRead + 1 to: lastRead + (data min: 20)); cr;
			nextPutAll: 'OutBuffer data size: ', (outNextToWrite - 1) asString; cr;
			nextPutAll: 'Out data (20):', (outBuffer copyFrom: 1 to: ((outNextToWrite - 1) min: 20)); cr]! !

!SocketStream methodsFor: 'printing' stamp: 'md 7/14/2006 12:28'!
print: anObject
	anObject printOn: self! !

!SocketStream methodsFor: 'printing' stamp: 'gk 2/10/2005 11:44'!
printOn: aStream
	"Display buffer sizes."

	aStream nextPutAll: self class name.
	inBuffer ifNotNil: [
		aStream nextPutAll: '[inbuf:',
		(inBuffer size / 1024) rounded asString, 'kb/outbuf:',
		(outBuffer size / 1024) rounded asString, 'kb]']! !

!SocketStream methodsFor: 'private-socket' stamp: 'gk 2/25/2005 14:20'!
receiveAvailableData
	"Receive available data (as much as fits in the inBuffer)
	but not waiting for more to arrive.
	Return the position in the buffer where the
	new data starts, regardless if anything
	was read, see #adjustInBuffer."
	
	recentlyRead := socket receiveAvailableDataInto: inBuffer startingAt: inNextToWrite.
	^self adjustInBuffer: recentlyRead! !

!SocketStream methodsFor: 'private-socket' stamp: 'gk 2/25/2005 14:20'!
receiveData
	"Receive data with timeout if it has been set.
	If shouldSignal is false we use the Socket methods
	that swallow those Exceptions, if it is true the
	caller will have to handle those Exceptions.
	Return the position in the buffer where the
	new data starts, regardless if anything
	was read, see #adjustInBuffer."
	
	recentlyRead := shouldSignal ifTrue: [
		self shouldTimeout ifTrue: [
				socket receiveDataSignallingTimeout: timeout
					into: inBuffer startingAt: inNextToWrite]
			ifFalse: [
				socket receiveDataSignallingClosedInto: inBuffer
					startingAt: inNextToWrite]]
				ifFalse: [
		self shouldTimeout ifTrue: [
			"This case is tricky, if it times out and is swallowed
			how does other methods calling this method repeatedly
			get to know that? And what should they do?"
				socket receiveDataTimeout: timeout
					into: inBuffer startingAt: inNextToWrite]
			ifFalse: [
				socket receiveDataInto: inBuffer
					startingAt: inNextToWrite]].
	^self adjustInBuffer: recentlyRead! !

!SocketStream methodsFor: 'private-socket' stamp: 'gk 2/9/2005 22:53'!
receiveDataIfAvailable
	"Only used to check that there really is data to read
	from the socket after it signals dataAvailable.
	It has been known to signal true and then still
	not have anything to read. See also isDataAvailable.
	Return the position in the buffer where the
	new data starts, regardless if anything
	was read, see #adjustInBuffer."

	recentlyRead := socket receiveSomeDataInto: inBuffer startingAt: inNextToWrite.
	^self adjustInBuffer: recentlyRead! !

SocketStream removeSelector: #buffered!
SocketStream removeSelector: #buffered:!
SocketStream removeSelector: #inStream!
SocketStream removeSelector: #outStream!
SocketStream removeSelector: #pushBack:!
SocketStream removeSelector: #resetInStream!
SocketStream removeSelector: #resetOutStream!
!classDefinition: #SocketStream category: #'Network-Kernel'!
Object subclass: #SocketStream
	instanceVariableNames: 'recentlyRead socket inBuffer outBuffer inNextToWrite outNextToWrite lastRead timeout autoFlush bufferSize binary shouldSignal'
	classVariableNames: ''
	poolDictionaries: ''
	category: 'Network-Kernel'!
PositionableStream removeSelector: #pushBack:!
"Postscript:
Leave the line above, and replace the rest of this comment by a useful one.
Executable statements should follow this comment, and should
be separated by periods, with no exclamation points (!!).
Be sure to put any further comments in double-quotes, like this one."
Socket initialize!

