We use a test_socketpair() function to wrap a few lines that test the socket's socketpair() function.
List 3.8 shows an example of socketpair, as follows:
#!/usr/bin/env python # Python Network Programming Cookbook, Second Edition -- Chapter - 3 # This program is optimized for Python 3.5.2. # It may run on any other version with/without modifications. # To make it run on Python 2.7.x, needs some changes due to API differences. # Follow the comments inline to make the program work with Python 2. import socket import os BUFSIZE = 1024 def test_socketpair(): """ Test Unix socketpair""" parent, child = socket.socketpair() pid = os.fork() try: if pid: print ("@Parent, sending message...") child.close() parent.sendall(bytes("Hello from parent!", 'utf-8')) # Comment out the above line and uncomment
the below line for Python 2.7. # parent.sendall("Hello from parent!") response = parent.recv(BUFSIZE) print ("Response from child:", response) parent.close() else: print ("@Child, waiting for message from parent") parent.close() message = child.recv(BUFSIZE) print ("Message from parent:", message) child.sendall(bytes("Hello from child!!", 'utf-8')) # Comment out the above line and
uncomment the below line for Python 2.7. # child.sendall("Hello from child!!") child.close() except Exception as err: print ("Error: %s" %err) if __name__ == '__main__': test_socketpair()
The output from the preceding script is as follows:
$ python 3_8_ipc_using_socketpairs.py @Parent, sending message... @Child, waiting for message from parent Message from parent: b'Hello from parent!' Response from child: b'Hello from child!!'