title: RabbitMQ connection issue
body: <p>I am totally new to rabbitmq and I am stuck while consuming the events from rabbitmq. I am trying to read from rabbitmq and create a data frame. after receiving every 1000 events I am writing my data frame to redshift which took around 2 min to complete.</p>
  <p>Now my problem is I am getting a channel closed error from pika after writing to redshift.</p>
  <p>The error I am getting is,</p>
  <blockquote>
  <p>2022-06-08 18:36:26:
  b'{&quot;message&quot;:{&quot;sku&quot;:&quot;JOY010&quot;,&quot;totalStock&quot;:17,&quot;distStock&quot;:{&quot;WH-CGK45&quot;:{&quot;qty&quot;:17},&quot;WH-SRG18&quot;:{&quot;qty&quot;:0},&quot;WH-SUB51&quot;:{&quot;qty&quot;:0},&quot;WH-MES07&quot;:{&quot;qty&quot;:0},&quot;WH-UPG04&quot;:{&quot;qty&quot;:0},&quot;WH-AAP02&quot;:{&quot;qty&quot;:0}}},&quot;timestamp&quot;:1654691691086}'
  2022-06-08 18:36:26: 1493 2022-06-08 18:36:26:
  b'{&quot;message&quot;:{&quot;sku&quot;:&quot;JOY012&quot;,&quot;totalStock&quot;:50,&quot;distStock&quot;:{&quot;WH-CGK45&quot;:{&quot;qty&quot;:50},&quot;WH-SRG18&quot;:{&quot;qty&quot;:0},&quot;WH-SUB51&quot;:{&quot;qty&quot;:0},&quot;WH-MES07&quot;:{&quot;qty&quot;:0},&quot;WH-UPG04&quot;:{&quot;qty&quot;:0},&quot;WH-AAP02&quot;:{&quot;qty&quot;:0}}},&quot;timestamp&quot;:1654691691174}'
  2022-06-08 18:36:26: connection_lost: StreamLostError: (&quot;Stream
  connection lost: SSLEOFError(8, 'EOF occurred in violation of protocol
  (_ssl.c:2548)')&quot;,) 2022-06-08 18:36:26: AMQP stack terminated, failed
  to connect, or aborted: opened=True, error-arg=StreamLostError:
  (&quot;Stream connection lost: SSLEOFError(8, 'EOF occurred in violation of
  protocol (_ssl.c:2548)')&quot;,); pending-error=None 2022-06-08 18:36:26:
  Stack terminated due to StreamLostError: (&quot;Stream connection lost:
  SSLEOFError(8, 'EOF occurred in violation of protocol
  (_ssl.c:2548)')&quot;,) 2022-06-08 18:36:26: Closing transport socket and
  unlinking: state=2; &lt;ssl.SSLSocket fd=10,
  family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0,
  laddr=('192.168.224.32', 55119)&gt; 2022-06-08 18:36:26: Unexpected
  connection close detected: StreamLostError: (&quot;Stream connection lost:
  SSLEOFError(8, 'EOF occurred in violation of protocol
  (_ssl.c:2548)')&quot;,) Traceback (most recent call last):   File
  &quot;/Users/simtiaz/Documents/Development/Development/inventory_snapshot.py&quot;,
  line 46, in consume_message
  for method_frame, properties, body in channel.consume(rabbitmq_queue, False, False, None, 1800):   File
  &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 1991, in consume
  self._process_data_events(time_limit=delta)   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 2026, in _process_data_events
  self.connection.process_data_events(time_limit=time_limit)   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 827, in process_data_events
  self._flush_output(timer.is_ready, common_terminator)   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 523, in _flush_output
  raise self._closed_result.value.error pika.exceptions.StreamLostError: Stream connection lost:
  SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:2548)')</p>
  <p>During handling of the above exception, another exception occurred:</p>
  <p>Traceback (most recent call last):   File
  &quot;/Users/simtiaz/Documents/Development/Development/inventory_snapshot.py&quot;,
  line 95, in consume_message
  requeued_messages = channel.cancel()   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 2083, in cancel
  self.basic_cancel(self._queue_consumer_generator.consumer_tag)   File
  &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 1756, in basic_cancel
  assert (consumer_info.cancelled_by_broker or AssertionError: ctag1.93ba887914d24c9c9782628d315a4bec</p>
  <p>During handling of the above exception, another exception occurred:</p>
  <p>Traceback (most recent call last):   File
  &quot;/Users/simtiaz/Documents/Development/Development/inventory_snapshot.py&quot;,
  line 104, in 
  consume_message()   File &quot;/Users/simtiaz/Documents/Development/Development/inventory_snapshot.py&quot;,
  line 98, in consume_message
  channel.close()   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/adapters/blocking_connection.py&quot;,
  line 1513, in close
  self._impl._raise_if_not_open()   File &quot;/Users/simtiaz/.conda/envs/Development/lib/python3.10/site-packages/pika/channel.py&quot;,
  line 1393, in _raise_if_not_open
  raise exceptions.ChannelWrongStateError('Channel is closed.') pika.exceptions.ChannelWrongStateError: Channel is closed.</p>
  </blockquote>
  <p>My code is</p>
  <pre><code>
      data = {}
      df_rabbit = pd.DataFrame(data, columns=[&quot;SNP_DATE&quot;, &quot;SNP_TIME&quot;, &quot;SKU_CODE&quot;, &quot;WAREHOUSE_CODE&quot;, &quot;INVENTORY&quot;])
  
      param = pika.URLParameters(rabbitmq_url)
      param._heartbeat = 1800
      param._blocked_connection_timeout = 1800
  
      connection = pika.BlockingConnection(param)
      channel = connection.channel()
  
      try:
  
          for method_frame, properties, body in channel.consume(rabbitmq_queue, False, False, None, 1800):
              logging.info(method_frame.delivery_tag)
              logging.info(body)
  
              json_msg = json.loads(body)
  
              SKU = json_msg[&quot;message&quot;][&quot;sku&quot;]
  
              epoch = datetime.fromtimestamp(json_msg[&quot;timestamp&quot;] / 1000)
              dt = epoch.strftime(&quot;%Y-%m-%d&quot;)
              tm = epoch.strftime(&quot;%H:%M:%S&quot;)
  
              warehouse = json_msg[&quot;message&quot;][&quot;distStock&quot;]
  
              val = []
              for i in warehouse:
                  w_code = i
                  w_qty = warehouse[i][&quot;qty&quot;]
  
                  val.append((str(dt), str(tm), str(SKU), str(w_code), str(w_qty)))
                  df_tmp = pd.DataFrame([[dt, tm, SKU, w_code, w_qty]],
                                        columns=[&quot;SNP_DATE&quot;, &quot;SNP_TIME&quot;, &quot;SKU_CODE&quot;, &quot;WAREHOUSE_CODE&quot;, &quot;INVENTORY&quot;])
                  df_rabbit = pd.concat([df_rabbit, df_tmp])
  
              # Acknowledge the message
              channel.basic_ack(method_frame.delivery_tag)
  
              # Write to DB after every 1000 events
              if method_frame.delivery_tag % 1000 == 0:
  
                  start = datetime.now()
                  logging.info(&quot;Writing to DB...&quot;)
                  df_rabbit.to_sql(&quot;inventory_snapshot_dev&quot;, conn, if_exists='append', index=False)
  
  
                  end = datetime.now()
                  logging.info(&quot;Time taken to write into DB: &quot; + str(end-start))
  
                  logging.info(&quot;Flushing dataframe......&quot;)
                  #df_rabbit.drop(df_rabbit.index, inplace=False)
                  df_rabbit = pd.DataFrame(data, columns=[&quot;SNP_DATE&quot;, &quot;SNP_TIME&quot;, &quot;SKU_CODE&quot;, &quot;WAREHOUSE_CODE&quot;, &quot;INVENTORY&quot;])
  
                  logging.info(&quot;Flushing Done.&quot;)
      except:
          requeued_messages = channel.cancel()
          logging.info('Requeued %i messages' % requeued_messages)
      finally:
          channel.close()
          connection.close()
          conn.close()```
  </code></pre>