本文整理匯總了Python中socket.send方法的典型用法代碼示例。如果您正苦于以下問題：Python socket.send方法的具體用法？Python socket.send怎么用？Python socket.send使用的例子？那么恭喜您, 這里精選的方法代碼示例或許可以為您提供幫助。您也可以進一步了解該方法所在模塊socket的用法示例。

在下文中一共展示了socket.send方法的24個代碼示例，這些例子默認根據受歡迎程度排序。您可以為喜歡或者感覺有用的代碼點贊，您的評價將有助于我們的系統推薦出更棒的Python代碼示例。

示例1: send

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def send(self, data):

"""

Send data via TCP service.

:param data: data

:type data: str

"""

try:

self.socket.send(data)

# TODO: rework logging to have LogRecord with extra=direction

# TODO: separate data sent/received from other log records ?

self._debug('> {}'.format(data))

except socket.error as serr:

if (serr.errno == 10054) or (serr.errno == 10053):

self._close_ignoring_exceptions()

info = "{} during send msg '{}'".format(serr.errno, data)

raise RemoteEndpointDisconnected('Socket error: ' + info)

else:

raise

開發者ID:nokia，項目名稱:moler，代碼行數:21，

示例2: __init__

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def __init__(self, port, data, exit_event, quiet=False):

"""

:param port: Destination port.

:param data: The data intend to be send in a loop.

:param exit_event: Exit event.

:param quiet: If true don't raise event when crash.

"""

Thread.__init__(self)

self.port = port

# FIXME: socket.send(data>>127998) without read blocks thread

if len(data) > 102400:

data = data[0:102400]

logging.error("Data is too long, using only first %d bytes",

len(data))

self.data = data

self.exitevent = exit_event

self.idx = 0

self.quiet = quiet

self.ret_code = 1 # sets to 0 when finish properly

開發者ID:avocado-framework，項目名稱:avocado-vt，代碼行數:21，

示例3: write_to_fd

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def write_to_fd(self, data: memoryview) -> int:

try:

return self.socket.send(data) # type: ignore

except ssl.SSLError as e:

if e.args[0] == ssl.SSL_ERROR_WANT_WRITE:

# In Python 3.5+, SSLSocket.send raises a WANT_WRITE error if

# the socket is not writeable; we need to transform this into

# an EWOULDBLOCK socket.error or a zero return value,

# either of which will be recognized by the caller of this

# method. Prior to Python 3.5, an unwriteable socket would

# simply return 0 bytes written.

return 0

raise

finally:

# Avoid keeping to data, which can be a memoryview.

# See https://github.com/tornadoweb/tornado/pull/2008

del data

開發者ID:opendevops-cn，項目名稱:opendevops，代碼行數:19，

示例4: run

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def run(self):

_ClientHandler.log_info("csocketsendThread(); socket.send thread start")

self._tx=None

while self.__threadrun==True:

try:

# get queue-value in blocking mode

self._tx=self._queue.get(True)

self._queue.task_done()

except:

self.__threadrun=False

_ClientHandler.log_critical("csocketsendThread();Error on queue.get()")

raise

try:

self._request.sendall(bytes(self._tx))

except:

self.__threadrun=False

_ClientHandler.log_critical("csocketsendThread();Error on socket.send")

raise

_ClientHandler.log_info("csocketsendThread(); socket.send thread terminated")

開發者ID:norberts1，項目名稱:hometop_HT3，代碼行數:23，

示例5: __waitfor_client_register

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def __waitfor_client_register(self):

self.request.settimeout(5)

try:

devicetypetmp=self.request.recv(20)

self._client_devicetype = devicetypetmp.decode('utf-8')

_ClientHandler.log_info("Client-ID:{0}; register(); got devicetype:{1}".format(self._myownID,self._client_devicetype))

#send client-ID to client

sendtemp=str(self._myownID)

self.request.sendall(sendtemp.encode("utf-8"))

except socket.timeout:

_ClientHandler.log_critical("Client-ID:{0}; Timeout occured, no devicetype was send".format(self._myownID))

raise

except socket.error as e:

# Something else happened, handle error, exit, etc.

_ClientHandler.log_critical("Client-ID:{0}; error '{1}' on socket.recv or socket.send".format(self._myownID, e))

raise

except Exception as e:

_ClientHandler.log_critical("Client-ID:{0}; unkown error '{1}'".format(self._myownID,e))

raise

finally:

self.request.settimeout(None)

開發者ID:norberts1，項目名稱:hometop_HT3，代碼行數:23，

示例6: write

?點贊 6

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def write(self, data, extra={}):

SMALL_PKT_BORDER = 0x7f

LARGE_PKT_BORGER = 256 ** 3

if len(data) % 4 != 0:

print_err("BUG: MTProtoFrameStreamWriter attempted to send msg with len", len(data))

return 0

if extra.get("SIMPLE_ACK"):

return self.upstream.write(data[::-1])

len_div_four = len(data) // 4

if len_div_four < SMALL_PKT_BORDER:

return self.upstream.write(bytes([len_div_four]) + data)

elif len_div_four < LARGE_PKT_BORGER:

return self.upstream.write(b'\x7f' + int.to_bytes(len_div_four, 3, 'little') + data)

else:

print_err("Attempted to send too large pkt len =", len(data))

return 0

開發者ID:alexbers，項目名稱:mtprotoproxy，代碼行數:22，

示例7: write_to_fd

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def write_to_fd(self, data):

return self.socket.send(data)

開發者ID:tao12345666333，項目名稱:tornado-zh，代碼行數:4，

示例8: __init__

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def __init__(self, moler_connection,

port, host="localhost", receive_buffer_size=64 * 4096,

logger=None):

"""Initialization of TCP-threaded connection."""

super(ThreadedTcp, self).__init__(port=port, host=host,

receive_buffer_size=receive_buffer_size,

logger=logger)

self.pulling_thread = None

# make Moler happy (3 requirements) :-)

self.moler_connection = moler_connection # (1)

self.moler_connection.how2send = self.send # (2)

開發者ID:nokia，項目名稱:moler，代碼行數:13，

示例9: testSendAfterClose

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def testSendAfterClose(self):

# testing send() after close() with timeout

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.settimeout(1)

sock.close()

self.assertRaises(socket.error, sock.send, "spam")

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:8，

示例10: _testRecv

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testRecv(self):

self.serv_conn.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例11: _testOverFlowRecv

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testOverFlowRecv(self):

self.serv_conn.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例12: _testRecvFrom

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testRecvFrom(self):

self.serv_conn.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例13: _testOverFlowRecvFrom

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testOverFlowRecvFrom(self):

self.serv_conn.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例14: _testDup

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testDup(self):

self.serv_conn.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例15: _testShutdown

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testShutdown(self):

self.serv_conn.send(MSG)

self.serv_conn.shutdown(2)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:5，

示例16: _testShutdown_overflow

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testShutdown_overflow(self):

import _testcapi

self.serv_conn.send(MSG)

# Issue 15989

self.assertRaises(OverflowError, self.serv_conn.shutdown,

_testcapi.INT_MAX + 1)

self.assertRaises(OverflowError, self.serv_conn.shutdown,

2 + (_testcapi.UINT_MAX + 1))

self.serv_conn.shutdown(2)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:11，

示例17: testSend

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def testSend(self):

self.serv.send(MSG)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:4，

示例18: send

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def send(self, data, flags=0):

n = self._sock.send(data, flags)

self.sent.append(data[:n])

return n

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:6，

示例19: testInsideTimeout

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def testInsideTimeout(self):

conn, addr = self.serv.accept()

self.addCleanup(conn.close)

time.sleep(3)

conn.send("done!")

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:7，

示例20: _testRecvIntoArray

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testRecvIntoArray(self):

with test_support.check_py3k_warnings():

buf = buffer(MSG)

self.serv_conn.send(buf)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:6，

示例21: _testRecvFromIntoArray

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testRecvFromIntoArray(self):

with test_support.check_py3k_warnings():

buf = buffer(MSG)

self.serv_conn.send(buf)

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:6，

示例22: _testStream

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def _testStream(self):

self.cli.send(MSG)

self.cli.close()

開發者ID:IronLanguages，項目名稱:ironpython2，代碼行數:5，

示例23: run_pool

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def run_pool():

try:

httpd = ThreadedTCPServer((State.config.get("proxy", "listen").strip(),

State.config.getint("proxy", "port")), Proxy)

except OSError as exc:

if "attempt was made" in str(exc):

print("Px failed to start - port in use")

else:

pprint(exc)

return

mainsock = httpd.socket

print_banner()

if hasattr(socket, "fromshare"):

workers = State.config.getint("settings", "workers")

for i in range(workers-1):

(pipeout, pipein) = multiprocessing.Pipe()

p = multiprocessing.Process(target=start_worker, args=(pipeout,))

p.daemon = True

p.start()

while p.pid is None:

time.sleep(1)

pipein.send(mainsock.share(p.pid))

serve_forever(httpd)

###

# Proxy detection

開發者ID:genotrance，項目名稱:px，代碼行數:32，代碼來源:px.py

示例24: run

?點贊 5

?

# 需要導入模塊: import socket [as 別名]

# 或者: from socket import send [as 別名]

def run(self):

logging.debug("ThSend %s: run", self.getName())

try:

while not self.exitevent.isSet():

self.idx += self.port.send(self.data)

logging.debug("ThSend %s: exit(%d)", self.getName(),

self.idx)

except Exception as ints:

if not self.quiet:

raise ints

logging.debug(ints)

self.ret_code = 0

開發者ID:avocado-framework，項目名稱:avocado-vt，代碼行數:14，

注：本文中的socket.send方法示例整理自Github/MSDocs等源碼及文檔管理平臺，相關代碼片段篩選自各路編程大神貢獻的開源項目，源碼版權歸原作者所有，傳播和使用請參考對應項目的License；未經允許，請勿轉載。

總結

以上是生活随笔為你收集整理的python socket send_Python socket.send方法代码示例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。