upstream/ipython Files · IPython/zmq/ktest.py

push() now swallows syntax errors and immediately produces a 'ready'...

push() now swallows syntax errors and immediately produces a 'ready' state. This is intended, as the invalid input may in fact be OK with ipython extensions, but that can only be ascertained by the kernel, so the client should send this right away to the kernel.

Brian Granger - - Load All Authors

File last commit:

r2606:f0587bad


                r2635:6112cd14

Download file

             ktest.py
        
                    51 lines
            
             | 1.1 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / zmq / ktest.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      from Queue import Queue, Empty

      import time

      from kernelmanager import KernelManager

      xreq_addr = ('127.0.0.1',5575)

      sub_addr = ('127.0.0.1', 5576)

      rep_addr = ('127.0.0.1', 5577)

      km = KernelManager(xreq_addr, sub_addr, rep_addr)

      # xreq_channel = km.get_xreq_channel()

      sub_channel = km.get_sub_channel()

      # xreq_channel.start()

      sub_channel.start()

      print "Channels are started"

      def printer(msg):

          print

          print msg

      class CallHandler(object):

          def __init__(self):

              self.queue = Queue()

          def __call__(self, handler, msg):

              self.queue.put((handler, msg))

          def handle(self):

              try:

                  handler, msg = self.queue.get(block=False)

              except Empty:

                  pass

              else:

                  handler(msg)

      call_handler = CallHandler()

      sub_channel.override_call_handler(call_handler)

      sub_channel.add_handler(printer, 'pyin')

      sub_channel.add_handler(printer, 'pyout')

      sub_channel.add_handler(printer, 'stdout')

      sub_channel.add_handler(printer, 'stderr')

      for i in range(100):

          call_handler.handle()

          time.sleep(1)

      # xreq_channel.join()

      sub_channel.join()

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				from Queue import Queue, Empty
				import time

				from kernelmanager import KernelManager

				xreq_addr = ('127.0.0.1',5575)
				sub_addr = ('127.0.0.1', 5576)
				rep_addr = ('127.0.0.1', 5577)


				km = KernelManager(xreq_addr, sub_addr, rep_addr)
				# xreq_channel = km.get_xreq_channel()
				sub_channel = km.get_sub_channel()

				# xreq_channel.start()
				sub_channel.start()

				print "Channels are started"

				def printer(msg):
				print
				print msg

				class CallHandler(object):

				def __init__(self):
				self.queue = Queue()

				def __call__(self, handler, msg):
				self.queue.put((handler, msg))

				def handle(self):
				try:
				handler, msg = self.queue.get(block=False)
				except Empty:
				pass
				else:
				handler(msg)

				call_handler = CallHandler()
				sub_channel.override_call_handler(call_handler)
				sub_channel.add_handler(printer, 'pyin')
				sub_channel.add_handler(printer, 'pyout')
				sub_channel.add_handler(printer, 'stdout')
				sub_channel.add_handler(printer, 'stderr')

				for i in range(100):
				call_handler.handle()
				time.sleep(1)

				# xreq_channel.join()
				sub_channel.join()